Legg til mulighet for kafka-header til statistikk (#1359)

navikt · Apr 12, 2024 · 27aa641 · 27aa641
1 parent e096146
commit 27aa641
Showing 1 changed file with 18 additions and 1 deletion.
diff --git a/...on/kafka-properties/src/main/java/no/nav/vedtak/felles/integrasjon/kafka/KafkaSender.java b/...on/kafka-properties/src/main/java/no/nav/vedtak/felles/integrasjon/kafka/KafkaSender.java
@@ -1,9 +1,12 @@
 package no.nav.vedtak.felles.integrasjon.kafka;
 
+import java.util.Optional;
+
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.header.internals.RecordHeader;
 
 import no.nav.vedtak.exception.IntegrasjonException;
 
@@ -21,16 +24,30 @@ public String getTopicName() {
         return topicName;
     }
 
+    public record KafkaHeader(String key, byte[] value) {}
+
     public RecordMetadata send(String key, String message) {
         if (topicName == null) {
             throw kafkaPubliseringException("null", new IllegalArgumentException());
         }
-        return send(key, message, this.topicName);
+        return send(null, key, message, topicName);
+    }
+
+    public RecordMetadata send(KafkaHeader header, String key, String message) {
+        if (topicName == null) {
+            throw kafkaPubliseringException("null", new IllegalArgumentException());
+        }
+        return send(header, key, message, this.topicName);
     }
 
     public RecordMetadata send(String key, String message, String topic) {
+        return send(null, key, message, topic);
+    }
+
+    public RecordMetadata send(KafkaHeader header, String key, String message, String topic) {
         try {
             var record = new ProducerRecord<>(topic, key, message);
+            Optional.ofNullable(header).ifPresent(h -> record.headers().add(new RecordHeader(h.key(), h.value())));
             return producer.send(record).get();
         } catch (Exception e) {
             if (e instanceof InterruptedException) {