Add support to read avro messages encoded with Confluent's kafkaavroserializer

[Kargreat]

I have avro messages encoded with kafkaavroserializer. I am not able to read them to elasticsearch using kafka input plugin.

I did classpath the kafka avro decoder (the entire serializer jars along with its dependencies) too. But I got an error that said that the plugin was not found.

do you happen to know this error or issue? do you have any thoughts to overcome this issue?

[joekiller]

@Kargreat Have you tried using the avro codec described in the following link? https://github.com/logstash-plugins/logstash-codec-avro/blob/master/DEVELOPER.md

Also check out https://github.com/logstash-plugins/logstash-codec-avro/blob/master/lib/logstash/codecs/avro.rb#L13-L17 to see how the schema is communicated.

[Kargreat]

@joekiller: I did read those posts and tried using logstadh-codec-avro in my input plugin. But those are for codecs that use binary decoder.

Also I went through their gits too. But they are different as we need to enhance the consumer with a different class. That corresponds to ruby-Kafka of yours.

Please let me know if you have any clarifications or queries.

[joekiller]

Okay I'll try to give this a more detailed look.
On Jul 8, 2015 1:10 PM, "Kargreat" notifications@github.com wrote:

@joekiller https://github.com/joekiller: I did read those posts and
tried using logstadh-codec-avro in my input plugin. But those are for
codecs that use binary decoder.

Also I went through their gits too. But they are different as we need to
enhance the consumer with a different class. That corresponds to ruby-Kafka
of yours.

Please let me know if you have any clarifications or queries.

—
Reply to this email directly or view it on GitHub
#37 (comment)
.

[talevy]

@Kargreat is this the serializer? https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroSerializer.java

This is not a regular Avro serialized message

[Kargreat]

@talevy: That is correct! That is the reason we need a consumer with a different decoder (the one for the KafkaAvroSerializer). The implementation can be found out: http://docs.confluent.io/1.0/schema-registry/docs/serializer-formatter.html & the KafkaAvroDecoder can be found at https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDecoder.java.

@joekiller: the links i have pasted should give you further more information.

[talevy]

@Kargreat cool! that is what I thought. This is a special serialization format with a subset of the message being the contents in Avro. A new codec with this implementation is needed beyond what the existing avro-codec can support

[Kargreat]

@tal: Will it be possible to enhance the codec to read values using the other decoder?

[talevy]

It is technically a separate serialization format, so would warrant a separate codec plugin.

[ilene-gorman]

If you have a customer deserializer, isn't that what the configuration / input/ kafka / decoder_class attribute is for?

When I set my decoder class to the Java class we use, I get an error that the class cannot be found, regardless of where I put the class, in the logstash main directory, or the logstash lib or bin directory. I've also put the jar file that contains the class file in all of those locations.

Can you please tell me after I've defined that class, how to use it in the logstash environment? What I'm trying to accomplish is to read avro data from kafka with a customized deserializer via logstash to elasticsearch.

Thanks for your help.

[suyograo]

In addition to the config changes, you'd need to add it to the classpath and start LS with your custom serializer.

input {
decoder_class => "com.mycomp.etl.avro.schema.kafka.RawLog2KafkaSerializer"
...
}

Logstash environment:

CLASSPATH=$CLASSPATH:/path/to/serializer.jar
bin/logstash -f <config_path>

I'll add this information to the docs

@ilene-gorman since you posted this on https://discuss.elastic.co/t/provide-custom-kafka-message-decoder-to-logstash/1780/3 lets follow up there.

[joekiller]

Finally RTFM'd this (http://docs.confluent.io/1.0/schema-registry/docs/serializer-formatter.html) and it seems pretty trivial to slip into jruby-kafka. I think I'll just pack Avro with the gem.

ignore this see below

[joekiller]

okay reading this more I think @talevy is correct that it just needs a codec. The input runs the raw message through the decoder prior to any mutations.

      @codec.decode("#{message_and_metadata.message}") do |event|
        decorate(event)
        if @decorate_events
          event['kafka'] = {'msg_size' => message_and_metadata.message.size,
                            'topic' => message_and_metadata.topic,
                            'consumer_group' => @group_id,
                            'partition' => message_and_metadata.partition,
                            'key' => message_and_metadata.key}
        end
        output_queue << event

The kafka plugin is just delivering raw binary by default via the decoder_class settings.

  # The serializer class for messages. The default decoder takes a byte[] and returns the same byte[]
  config :decoder_class, :validate => :string, :default => 'kafka.serializer.DefaultDecoder'

So if I make a codec that does the Kafka Avro Decoder, should be good to go.