From 078da23910fd986dfd0243ddb6110b142391db78 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 7 Apr 2025 22:15:12 +0200 Subject: [PATCH] =?utf8?q?ROT:=20Erwartungen=20f=C3=BCr=20Standard-Header?= =?utf8?q?=20definiert?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- src/main/java/de/juplo/kafka/DeadLetterConsumer.java | 9 +++++++++ src/test/java/de/juplo/kafka/ApplicationTests.java | 7 +++++++ 2 files changed, 16 insertions(+) diff --git a/src/main/java/de/juplo/kafka/DeadLetterConsumer.java b/src/main/java/de/juplo/kafka/DeadLetterConsumer.java index 113fdc2b..9eb2f8be 100644 --- a/src/main/java/de/juplo/kafka/DeadLetterConsumer.java +++ b/src/main/java/de/juplo/kafka/DeadLetterConsumer.java @@ -25,6 +25,10 @@ import java.util.stream.IntStream; @Slf4j public class DeadLetterConsumer implements Runnable, SmartLifecycle { + public final static String KEY = "KEY"; + public final static String TIMESTAMP = "TIMESTAMP"; + + private final String id; private final String topic; private final int numPartitions; @@ -234,6 +238,11 @@ public class DeadLetterConsumer implements Runnable, SmartLifecycle return Mono.fromFuture(future); } + String prefixed(String headerName) + { + return headerName; + } + @Override public boolean isRunning() { diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index 0d9b45a9..7b8f3a6a 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -26,6 +26,7 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.context.EmbeddedKafka; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; @@ -72,6 +73,10 @@ public class ApplicationTests recordMetadata.offset()); assertThat(response.getStatusCode()) .isEqualTo(HttpStatusCode.valueOf(HttpStatus.OK.value())); + assertThat(response.getHeaders().getValuesAsList(DeadLetterConsumer.KEY)) + .containsExactly(key); + assertThat(response.getHeaders().getValuesAsList(DeadLetterConsumer.TIMESTAMP)) + .containsExactly(Long.toString(recordMetadata.timestamp())); assertThat(response.getBody()) .isEqualTo(value); } @@ -123,6 +128,8 @@ public class ApplicationTests AdminClient adminClient; @Autowired TestRestTemplate restTemplate; + @Autowired + DeadLetterConsumer deadLetterConsumer; final long[] currentOffsets = new long[NUM_PARTITIONS]; -- 2.39.5