From 2a63ef32b52c7cd8a8d997a09ab978366184aac6 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Mon, 7 Apr 2025 23:43:02 +0200
Subject: [PATCH] =?utf8?q?ROT:=20Erwartungen=20f=C3=BCr=20User-Header=20de?=
 =?utf8?q?finiert?=
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

---
 .../java/de/juplo/kafka/ApplicationTests.java | 21 ++++++++++++++++++-
 1 file changed, 20 insertions(+), 1 deletion(-)

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 51beb455..824c6142 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -4,6 +4,7 @@ import com.jayway.jsonpath.JsonPath;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.AfterEach;
@@ -27,7 +28,9 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.web.util.UriUtils;
 
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -119,6 +122,12 @@ public class ApplicationTests
         Long.toString(recordMetadata.timestamp()));
     assertThat(response.getBody())
       .isEqualTo(value);
+
+    result.getProducerRecord().headers().forEach(header ->
+      assertThat(response.getHeaders().toSingleValueMap())
+        .containsEntry(
+          UriUtils.encodePathSegment(HEADER_PREFIX + header.key(), StandardCharsets.UTF_8),
+          UriUtils.encodePathSegment(new String(header.value()), StandardCharsets.UTF_8)));
   }
 
   @DisplayName("Not yet existing offset")
@@ -213,8 +222,18 @@ public class ApplicationTests
   {
     String key = Integer.toString(partition);
     String value = "Hällö Wöhrld!%? -- " + ++message;
+    ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
+      TOPIC,
+      partition,
+      key.getBytes(),
+      value.getBytes());
+    record
+      .headers()
+      .add("p%" + partition, "!§$%&/()ß#~üöä<;:|@€¢".getBytes())
+      .add(Integer.toString(message), "\"".getBytes())
+      .add("€", ":".getBytes());
     return kafkaTemplate
-      .send(TOPIC, partition, key.getBytes(), value.getBytes())
+      .send(record)
       .thenApply(result ->
       {
         RecordMetadata metadata = result.getRecordMetadata();
-- 
2.39.5