Payload deserialization fails when basic auth is used with schema registry

[robbavey]

Steps to reproduce:

• Setup confluent schema registry using the same version included in the integration tests in this plugin
• Add the following lines to schema-registry.properties:

authentication.method=BASIC
authentication.roles=admin,developer,user,sr-user
authentication.realm=SchemaRegistry-Props

• Create a jaas.config file to enable password login on the schema registry:

SchemaRegistry-Props {
  org.eclipse.jetty.jaas.spi.PropertyFileLoginModule required
  file="/users/robbavey/code/plugins/logstash-integration-kafka/pwd"
  debug="true";
};

• Create a password file to authenticate against:

barney: changeme,user,developer

• Start schema registry pointing to jaas.config
• I used an amended version of the test at https://github.com/logstash-plugins/logstash-integration-kafka/blob/master/spec/integration/inputs/kafka_spec.rb#L362 to write data into the protected topic
• Start logstash with the following config:

input {
kafka {
'schema_registry_url' => 'http://localhost:8081'
'schema_registry_key' => 'barney'
'schema_registry_secret' => 'changeme'
'topics' => ['authed']
'codec' => 'plain'
'group_id' => 'xxxx'
'auto_offset_reset' => 'earliest'
 'enable_auto_commit' => 'true' }
}

Observed Behaviour
Once data starts flowing into the topic, the following error occurs:

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition authed-0 at offset 0. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro unknown schema for id 81
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Unauthorized; error code: 401

[robbavey]

Fixed by #94