인프런 커뮤니티 질문&답변

현상:

아래에도 동일하게 이슈 올려주신 분이 있는 것으로 보이는데 postman으로 다음과 같이 요청하였을 때, kafka에서 DLT 소비하는 과정에서 무한 반복하는 현상이 발생하면서 DLT 토픽 자체에 계속 쌓이게 됩니다!

@RetryableTopic에 autoStartDltHandler = False 설정을 해주거나, @DLTHandler를 정의 해줘야 DLT 소비를 무한으로 하지 않는 것 같은데 뭔가 설정이 잘못 된게 있을까요? ㅠ

요청 파라미터:

{
    "from": "sender@naver.com", 
    "to": "fail@naver.com",
    "subject": "heyyyy",
    "body": "helloo"
}

2026-02-01T15:53:25.480+09:00 ERROR 15124 --- [email-send-consumer] [ner#0.dlt-0-C-1] k.r.DeadLetterPublishingRecovererFactory : Record: topic = email.send.dlt, partition = 0, offset = 605, main topic = email.send threw an error at topic email.send.dlt and won't be retried. Sending to DLT with name email.send.dlt.

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:3014) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2914) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2880) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2790) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2632) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2526) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2164) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1541) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1479) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1348) ~[spring-kafka-4.0.2.jar:4.0.2]
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[na:na]
	at java.base/java.lang.Thread.run(Thread.java:1583) ~[na:na]
Caused by: org.springframework.kafka.listener.TimestampedException: Exception thrown at 2026-02-01T06:53:25.480044400Z
	at org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter.onMessage(KafkaBackoffAwareMessageListenerAdapter.java:100) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter.onMessage(KafkaBackoffAwareMessageListenerAdapter.java:49) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2901) ~[spring-kafka-4.0.2.jar:4.0.2]
	... 10 common frames omitted
Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: invokeHandler Failed
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.checkAckArg(MessagingMessageListenerAdapter.java:519) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:496) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:425) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter.invokeDelegateOnMessage(KafkaBackoffAwareMessageListenerAdapter.java:106) ~[spring-kafka-4.0.2.jar:4.0.2]
	at org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter.onMessage(KafkaBackoffAwareMessageListenerAdapter.java:97) ~[spring-kafka-4.0.2.jar:4.0.2]
	... 12 common frames omitted
Caused by: java.lang.IllegalStateException: No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.
	... 19 common frames omitted

Caused by: org.springframework.kafka.listener.TimestampedException: Exception thrown at 2026-02-01T06:53:25.480044400Z

Caused by: org.springframework.kafka.listener.ListenerExecutionFailedException: invokeHandler Failed

Caused by: java.lang.IllegalStateException: No Acknowledgment available as an argument, the listener container must have a MANUAL AckMode to populate the Acknowledgment.


> Task :com.example.emailsendconsumer.EmailSendConsumerApplication.main() FAILED

Execution failed for task ':com.example.emailsendconsumer.EmailSendConsumerApplication.main()'.
> Build cancelled while executing task ':com.example.emailsendconsumer.EmailSendConsumerApplication.main()'

* Try:
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Run with --scan to get full insights from a Build Scan (powered by Develocity).
> Get more help at https://help.gradle.org.
BUILD FAILED in 17s
3 actionable tasks: 2 executed, 1 up-to-date

코드 발생:

오류 발생하는 것으로 보이는 consumer 코드는 다음과 같습니다.

우선 강사님 코드와 다른 점은 Backoff 부분인데 이건 라이브러리 버전 문제 인 것 같아서 넘어갔습니다.

@Service
public class EmailSendConsumer {

  @KafkaListener(
      topics = "email.send",
      groupId = "email-send-group"
  )
  @RetryableTopic(
      attempts = "5",
      backOff = @org.springframework.kafka.annotation.BackOff(
          delay = 1000,
          multiplier = 2.0
        ),
      dltTopicSuffix = ".dlt"
  )
  public void consume(String message) {
    System.out.println("Received message: " + message);
    EmailSendMessage emailSendMessage = EmailSendMessage.fromJson(message);

    if (emailSendMessage.getTo().equals("fail@naver.com")) {
      // 잘못된 이메일 주소로 인해 발송 실패
      throw new RuntimeException("Simulated email sending failure");
    }

    // 이메일 전송 로직 구현 생략 (예: SMTP 서버를 통해 이메일 전송)
    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      throw new RuntimeException("Email sending interrupted", e);
    }

    System.out.println("Email sent to: " + emailSendMessage.getTo());
  }
}