add key/value decoder options to Group for dynamically using custom decoder for consuming kafka messages

[talevy]

I am interested in adding config params to logstash-input-kafka so that people are able to apply their own decoders to either the message itself (value) or the message's key.

example config:

input {
  kafka {
    topic_id => "my_avro"
    value_decoder_class => "org.personal.KafkaAvroDecoder"
  }
}

Unfortunately, this information will be passed in at the runtime of the plugin, so we need to do some reflection. This strategy leads to a few assumptions made about the constructors of all Decoders that are chosen. Currently, the kafka.serializer.DefaultDecoder's only constructor is one with a VerifiableProperties argument. Given this, all other Decoders are assumped to have the same, even if this is unnecessary. Here is an example custom decoder for the sake of prototyping: DotsDecoder

@joekiller, this is very much a WIP and any suggestions are welcome!

[talevy]

@joekiller have time to check this out?

[joekiller]

Nice, looks straightforward. Have you tested it?

[talevy]

yes, I have only tested it via https://github.com/logstash-plugins/logstash-input-kafka/pull/29

but it is kind of testing it on its own by replacing how the DefaultDecoder gets loaded in the default case.

Given your response, I feel comfortable replacing the WIP to a valid PR :)

[talevy]

One thing I should note is that the current version of this PR will only support Decoders that do not utilize any VerifiableProperties because I hardcoded the nil.

that being said. There is no way, currently, to provide your own Properties argument in the Encoder in the Producer, unless I am mistaken.

[joekiller]

This looks fine to me. The only codec that I could find that uses the properties is the default encoder. Eventually we can just create a java properties from a map of properties for each codec but the null is fine for now.

[talevy]

thats what I figure. sweet!