X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=delivery%2Fsrc%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Foutbox%2Fdelivery%2FOutboxProducer.java;h=29d827acde5aed9bd9d07ce3276ac521759d721c;hb=724def8b418ad48dcc2838b01058cf12ccb50a55;hp=ac7d4ebff2cbcd44280b469f84bb87805f0d670c;hpb=e3865636a1236c681127a4d267ed5bbf334a992b;p=demos%2Fkafka%2Foutbox

diff --git a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
index ac7d4eb..29d827a 100644
--- a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
+++ b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
@@ -17,11 +17,15 @@ import org.springframework.scheduling.annotation.Scheduled;
 
 import javax.annotation.PreDestroy;
 
+import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.*;
+
 
 public class OutboxProducer
 {
   final static Logger LOG = LoggerFactory.getLogger(OutboxProducer.class);
 
+  public final static String HEADER = "#";
 
   private final OutboxRepository repository;
   private final KafkaProducer<String, String> producer;
@@ -41,9 +45,10 @@ public class OutboxProducer
     this.repository = repository;
 
     Properties props = new Properties();
-    props.put("bootstrap.servers", properties.bootstrapServers);
-    props.put("key.serializer", StringSerializer.class.getName());
-    props.put("value.serializer", StringSerializer.class.getName());
+    props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
+    props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+    props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+    props.put(ENABLE_IDEMPOTENCE_CONFIG, true);
 
     this.producer = new KafkaProducer<>(props);
     this.topic = properties.topic;
@@ -83,7 +88,7 @@ public class OutboxProducer
         new ProducerRecord<>(topic, item.getKey(), item.getValue());
 
     sequenceNumber = item.getSequenceNumber();
-    record.headers().add("SEQ#", Longs.toByteArray(sequenceNumber));
+    record.headers().add(HEADER, Longs.toByteArray(sequenceNumber));
 
     producer.send(record, (metadata, e) ->
     {