[Feature Request]: Allow reading of non-string messages with a Kafka Consumer

[hansva]

What would you like to happen?

Migration: https://issues.apache.org/jira/browse/HOP-3687

For both the regular Kafka Consumer and the Beam Kafka Consumer transforms it would be great to be able to read non-String messages, for example in Avro format.

For the Beam Kafka Consumer I think it would be cool if you could specify the output format as "Avro Record" (in our example) or Serializable, Binary, with String as a default. 

This would allow us to set Kafka consumer option value.deserializer to KafkaAvroDeserializer and get an actual Avro Record as a message.

Issue Priority

Priority: 3

Issue Component

Component: Transforms

[vbhanuchander-lang]

I was looking at picking this up, but I think it may already be implemented and the issue can probably be closed.

KafkaConsumerField.Type currently declares String, Integer, Number, Binary and Avro, each carrying its serializer and deserializer:

Binary("Binary", IValueMeta.TYPE_BINARY,
       ByteArraySerializer.class.getName(), ByteArrayDeserializer.class.getName()),
Avro("Avro", IValueMeta.TYPE_AVRO,
     "io.confluent.kafka.serializers.KafkaAvroSerializer",
     "io.confluent.kafka.serializers.KafkaAvroDeserializer");

KafkaFactory.consumer(...) applies the selected type to ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG and KEY_DESERIALIZER_CLASS_CONFIG, the consumer dialog offers every value of that enum for the output fields, and io.confluent:kafka-avro-serializer plus kafka-schema-registry-client are dependencies of plugins/transforms/kafka. The schema registry URL can be supplied through the existing advanced options table.

Going by git log, this landed in e9921c8aba (2022-02-08) under "HOP-3739 : Add support for Avro in Kafka transforms". This issue is HOP-3687, which was migrated to GitHub in March 2023 — so it looks like the sibling ticket implemented it and the migrated copy was never closed.

If someone can confirm, this could be closed as already done. If there is a specific part still missing — Beam Kafka Consumer rather than the regular transform, or a "Serializable" output option — I'm happy to pick that up instead; I've been working in these transforms recently (#7747).