add headers to native producer

sauljabin · Jul 12, 2024 · e048445 · e048445
1 parent 1b2b677
commit e048445
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/kafka-avro-clients/src/main/java/kafka/sandbox/cli/Producer.java b/kafka-avro-clients/src/main/java/kafka/sandbox/cli/Producer.java
@@ -44,7 +44,7 @@ public Integer call() {
             Supplier supplier = newMessage();
             ProducerRecord<String, Supplier> record = new ProducerRecord<>(
                     topic,
-                    supplier.getId().toString(),
+                    supplier.getId(),
                     supplier
             );
             producer.send(

diff --git a/kafka-native-clients/src/main/java/kafka/sandbox/cli/Producer.java b/kafka-native-clients/src/main/java/kafka/sandbox/cli/Producer.java
@@ -5,8 +5,10 @@
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.serialization.Serializer;
 
+import java.util.List;
 import java.util.Properties;
 
 @Slf4j
@@ -25,8 +27,13 @@ public void produce(String topic, int message) {
             V value = newMessage();
             ProducerRecord<String, V> record = new ProducerRecord<>(
                     topic,
+                    null,
                     String.valueOf(value),
-                    value
+                    value,
+                    List.of(
+                            new RecordHeader("native", value.getClass().getName().getBytes()),
+                            new RecordHeader("value", String.valueOf(value).getBytes())
+                    )
             );
             producer.send(
                     record,