X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FDeadLetterTopicConsumer.java;fp=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FDeadLetterTopicConsumer.java;h=0000000000000000000000000000000000000000;hb=25c2044064722af20f64651a32e94fb392710bbc;hp=ac8c65de2011e1491bad0885ecfdf213b4a578e2;hpb=66ff7d205e66616de8aaca94503dbbcd7d281f6d;p=demos%2Fkafka%2Ftraining

diff --git a/src/test/java/de/juplo/kafka/DeadLetterTopicConsumer.java b/src/test/java/de/juplo/kafka/DeadLetterTopicConsumer.java
deleted file mode 100644
index ac8c65d..0000000
--- a/src/test/java/de/juplo/kafka/DeadLetterTopicConsumer.java
+++ /dev/null
@@ -1,40 +0,0 @@
-package de.juplo.kafka;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.kafka.annotation.KafkaListener;
-import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.messaging.Message;
-
-import java.nio.ByteBuffer;
-import java.util.LinkedList;
-import java.util.List;
-
-
-@Slf4j
-public class DeadLetterTopicConsumer
-{
-  List<Message<?>> messages = new LinkedList<>();
-
-
-  @KafkaListener(
-      id = "DLT",
-      topics = "${sumup.adder.topic}.DLT",
-      containerFactory = "dltContainerFactory")
-  public void receive(Message<?> message)
-  {
-    log.info(
-        "Received dead letter on {}-{}|{},{} for {}-{}|{},{}: {}, exception={}",
-        message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC),
-        message.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID),
-        message.getHeaders().get(KafkaHeaders.OFFSET),
-        message.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY),
-        new String(message.getHeaders().get(KafkaHeaders.DLT_ORIGINAL_TOPIC, byte[].class)),
-        ByteBuffer.wrap(message.getHeaders().get(KafkaHeaders.DLT_ORIGINAL_PARTITION, byte[].class)).getInt(),
-        ByteBuffer.wrap(message.getHeaders().get(KafkaHeaders.DLT_ORIGINAL_OFFSET, byte[].class)).getLong(),
-        message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
-        message.getPayload(),
-        new String(message.getHeaders().get(KafkaHeaders.DLT_EXCEPTION_MESSAGE, byte[].class)));
-
-    messages.add(message);
-  }
-}