Spring云流kafka使用者错误处理和重试问题

Question

我需要在Spring云流kafka活页夹错误处理场景的帮助。我的应用程序有一个java 8消费者，其绑定在application.yaml中指定。

@Bean
public Consumer<Message<Transaction>> doProcess() {

    return message -> {
        Transaction transaction = message.getPayload();
       
        if(true) {
            throw new RuntimeException("exception!! !!:)");
        }
       Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, 
       Acknowledgment.class);
       if (acknowledgment != null) {
           System.out.println("Acknowledgment provided");
           acknowledgment.acknowledge();
       }
  }
}

application.yaml：

spring.application.name: appname
spring.cloud.stream:
  function.definition: doProcess
  kafka:
    default.consumer:
      startOffset: latest
      useNativeDecoding: true
    bindings:
      input.consumer.autoCommitOffset: false

bindings:
  doProcess-in-0:
    destination: kafka.input.topic.name
    group: appGroup
    content-type: application/*+avro
    consumer:
      autoCommitOffset: false.

现在，我正在处理错误，有两个问题:

>

我正在尝试手动包装消息的消费，而不是使用自动提交偏移设置为真。因此，当我将自动提交偏移设置设为假并测试错误场景时，会面临奇怪的行为，即每当抛出异常时，消息都会重试n次，即使在服务重新启动后也会发生失败消息的重试/重新传递（如果在n次重试完成之前重新启动）。一旦n次重试完成，即使在服务重新启动后也不会选择消息。那么这是否意味着，消费者在n次重试/重新传递消息后提交偏移量，这不应该是这种情况，因为自动提交偏移是错误的。

注意：我尚未配置任何 dlq。

我们需要编写自定义异常处理程序，在那里我们可以捕获异常（应用程序代码和框架中的错误）并通过AWS环境中的电子邮件向用户组发送通知。但是，我们无法找到任何可以捕获这两种类型异常的错误处理程序。比如扩展SeekTo货币错误处理程序或任何其他可以在错误事件上调用的侦听器。

编辑：

根据 Gary 提供的解决方案，我们可以使用以下 bean 来配置自定义错误处理程序：

@Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer> MQLCC() {
        System.out.println(String.format("DEBUG: Bean %s has bean created.", "MQLCC"));
        return new ListenerContainerCustomizerCustom ();
    }

    private static class ListenerContainerCustomizerCustom implements ListenerContainerCustomizer<AbstractMessageListenerContainer> {
        @Override
        public void configure(AbstractMessageListenerContainer container, String destinationName, String group) {
            System.out.println(String.format("HELLO from container %s, destination: %s, group: %s", container, destinationName, group));
        }

    }

Answer 1

侦听器容器中的默认错误处理程序将重试10次，然后记录错误并丢弃记录；对于不同的行为，您需要配置自定义错误处理程序和恢复策略。使用ListenerContainerCustomizerbean来配置容器。

请参阅https://docs . spring . io/spring-Kafka/docs/current/reference/html/# default-eh和https://docs . spring . io/spring-Kafka/docs/current/reference/html/# dead-letters

（3.2及更高版本）

或https://docs.spring.io/spring-kafka/docs/2.7.x/reference/html/#seek-至当前和https://docs.spring.io/spring-kafka/docs/2.7.x/reference/html/#dead-字母

对于早期版本。