Implementera dead-letter-topics (DLT)

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@SpringBootApplication
public class DltApplication {

  public static void main(String[] args) {
    SpringApplication.run(DltApplication.class, args);
  }

  @Bean
  public DefaultErrorHandler errorHandler(KafkaTemplate<?, ?> kafkaTemplate) {
    // Publish to DLT after 3 delivery attempts
    // FixedBackOff(interval, maxAttempts) -> interval is ignored for DLT after retries
    return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(kafkaTemplate),
                                   new FixedBackOff(0L, 2L)); // 0L interval, 2 retries = 3 attempts total
  }

  @KafkaListener(topics = "my-main-topic", groupId = "my-group", errorHandler = "errorHandler")
  public void listen(String message) {
    System.out.println("Received: " + message);
    if (message.contains("fail")) {
      throw new RuntimeException("Simulating processing failure!");
    }
  }

  // To send messages for testing (not part of DLT config itself)
  // @Autowired
  // private KafkaTemplate<String, String> template;
  // @EventListener(ApplicationReadyEvent.class)
  // public void sendMessage() {
  //   template.send("my-main-topic", "Hello");
  //   template.send("my-main-topic", "This will fail");
  // }
}

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.backoff.FixedBackOff;

@SpringBootApplication
public class CustomDltApplication {

  public static void main(String[] args) {
    SpringApplication.run(CustomDltApplication.class, args);
  }

  @Bean
  public DefaultErrorHandler customErrorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
    DeadLetterPublishingRecoverer customRecoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
      (record, exception) -> { // Custom DLT topic/partition resolver
        System.out.println("Sending to DLT: " + record.topic() + ".custom.dlt");
        return new DeadLetterPublishingRecoverer.HeaderNames(record.topic() + ".custom.dlt", null);
      });
    return new DefaultErrorHandler(customRecoverer, new FixedBackOff(0L, 1L)); // 1 retry = 2 attempts total
  }

  @KafkaListener(topics = "another-topic", groupId = "my-custom-group", errorHandler = "customErrorHandler")
  public void listenWithCustomDlt(String message) {
    System.out.println("Received (custom DLT): " + message);
    if (message.contains("fail")) {
      throw new RuntimeException("Simulating custom DLT failure!");
    }
  }
}

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.util.backoff.FixedBackOff;

@SpringBootApplication
public class DltConsumerApplication {

  public static void main(String[] args) {
    SpringApplication.run(DltConsumerApplication.class, args);
  }

  @Bean
  public DefaultErrorHandler errorHandler(KafkaTemplate<?, ?> kafkaTemplate) {
    return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(kafkaTemplate),
                                   new FixedBackOff(0L, 2L));
  }

  @KafkaListener(topics = "my-main-topic", groupId = "my-group", errorHandler = "errorHandler")
  public void listenMain(String message) {
    System.out.println("Main Listener Received: " + message);
    if (message.contains("fail")) {
      throw new RuntimeException("Main processing failure!");
    }
  }

  @KafkaListener(topics = "my-main-topic.DLT", groupId = "dlt-group")
  public void listenDlt(String message,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic,
                        @Header(KafkaHeaders.ORIGINAL_OFFSET) Long originalOffset,
                        @Header(KafkaHeaders.EXCEPTION_MESSAGE) String exceptionMessage) {
    System.out.println("DLT Listener Received: " + message);
    System.out.println("  From Topic: " + receivedTopic);
    System.out.println("  Original Offset: " + originalOffset);
    System.out.println("  Exception: " + exceptionMessage);
    // Here you would implement logic to store, alert, or reprocess
  }
}