From: Kai Moritz <kai@juplo.de>
Date: Mon, 7 Apr 2025 21:43:02 +0000 (+0200)
Subject: ROT: Erwartungen für User-Header definiert
X-Git-Tag: consumer/nodlt--2026-03-22--22-01~3
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=2a63ef32b52c7cd8a8d997a09ab978366184aac6;p=demos%2Fkafka%2Ftraining

ROT: Erwartungen für User-Header definiert
---

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();