Spring Kafka SeekToCurrentErrorHandler查找哪个记录失败

Question

我已经用kafkahandler实现了一个Kafka消费者。我的使用者应该使用事件，然后为每个事件向其他服务发送REST请求。只有当REST服务关闭时，我才想重试。否则，我可以忽略失败的事件。

我的容器工厂配置如下：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, MyCustomEvent>
  kafkaListenerContainerFactory() {

  ConcurrentKafkaListenerContainerFactory<String, MyCustomEvent> factory =
    new ConcurrentKafkaListenerContainerFactory<>();

  factory.setConsumerFactory(consumerFactory());
  factory.setStatefulRetry(true);
  factory.setRetryTemplate(retryTemplate());
  factory.setConcurrency(3);

  ContainerProperties containerProperties = factory.getContainerProperties();
  containerProperties.setAckOnError(false);
  containerProperties.setAckMode(AckMode.RECORD);
  containerProperties.setErrorHandler(new SeekToCurrentErrorHandler());

  return factory;
}

我使用ExceptionClassifierRetryPolicy来设置异常和相应的重试策略。

如果我有机会了解SeekToCurrEnterrorHandler中的哪个记录失败了，那么我将创建SeekToCurrEnterrorHandler的自定义实现，以检查失败的消息是否可重试（通过使用ThrownException字段）。如果它是不可重试的，那么我将从记录列表中删除它以重新查找。

关于如何实现这个功能有什么想法吗？

注意:enable.auto.commit设置为false，auto.offset.reset设置为最早。

谢谢你！

Answer 1

Apache Kafka2.2（尚未发布）的FailedRecordTracker:

https://docs.spring.io/spring-kafka/docs/2.2.0.m2/reference/html/whats-new-part.html#_listener_container_changes

从2.2版开始，SeekToCurrEnterrorHandler现在可以恢复（跳过）一个持续失败的记录。默认情况下，在10次失败后，将记录失败的记录（错误）。您可以使用自定义恢复器(biconsumer)和/或最大故障来配置处理程序。

SeekToCurrentErrorHandler errorHandler =
    new SeekToCurrentErrorHandler((record, exception) -> {
          // recover after 3 failures - e.g. send to a dead-letter topic
          }, 3);

因此，您所需要的只是将FailedRecordTracker和SeekToCurrEnterRorHandler源代码从主复制/粘贴到项目中，您就会拥有您正在寻找的功能：

https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/failedrecordtracker.java

https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/seektocurrenterrorhandler.java