Merge branch 'master' into feature/TFP-5750

Author: jolarsen

integrasjon/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) {