X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FRestProducer.java;h=b430e35f99911af6668d2d64f6155e710e674001;hb=0d99ee9ce803ea239c6d7746a87eb52ccef99f26;hp=59d2c77e8da56e8a4325ab88233f8bf6e61de158;hpb=1b978296e798614b3ca8317b43acd1a44a774ecd;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/RestProducer.java b/src/main/java/de/juplo/kafka/RestProducer.java
index 59d2c77..b430e35 100644
--- a/src/main/java/de/juplo/kafka/RestProducer.java
+++ b/src/main/java/de/juplo/kafka/RestProducer.java
@@ -9,6 +9,7 @@ import org.springframework.web.bind.annotation.*;
 import org.springframework.web.context.request.async.DeferredResult;
 
 import javax.annotation.PreDestroy;
+import java.math.BigInteger;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 
@@ -57,10 +58,17 @@ public class RestProducer
 
     final ProducerRecord<String, String> record = new ProducerRecord<>(
         topic,  // Topic
+        partition, // Partition
         key,    // Key
         value   // Value
     );
 
+    record.headers().add("source", id.getBytes());
+    if (correlationId != null)
+    {
+      record.headers().add("id", BigInteger.valueOf(correlationId).toByteArray());
+    }
+
     producer.send(record, (metadata, e) ->
     {
       long now = System.currentTimeMillis();