Publish key from kafka source as key to kafka sink / HyperdriveContext

[kevinwallimann]

Currently, the key from a kafka source is not ingested in ConfluentAvroKafkaStreamDecoder. Also, no key is produced to the kafka sink in KafkaStreamWriter. It should be possible to publish the ingested keys along with the value.

Consuming keys is supported by Abris like so

val result: DataFrame  = dataFrame.select(
    from_confluent_avro(col("key"), keyRegistryConfig) as 'key,
    from_confluent_avro(col("value"), valueRegistryConfig) as 'value)

Tasks

• Add config properties for decoder

  • decoder.avro.consume.keys (true / false)
  • decoder.avro.key.schema.naming.strategy
  • decoder.avro.key.schema.id
  • decoder.avro.key.schema.record.name
  • decoder.avro.key.schema.record.namespace

• Add config properties for writer

  • writer.kafka.produce.keys (true / false)
  • writer.kafka.key.schema.naming.strategy
  • writer.kafka.key.schema.record.name
  • writer.kafka.key.schema.record.namespace

• Add context object (HyperdriveContext) to which a value can be put and retrieved by key

  • To avoid key collisions, used keys should be documented
  • ConfluentAvroKafkaStreamDecoder should put the key column names, KafkaStreamWriter should retrieve them