SpringKafka搜索到当前错误处理程序最大故障在并发级别小于分区数时不起作用

Question

我使用的是spring kafka 2.2.7，我的消费者配置代码如下:

@Slf4j
@Configuration
@EnableKafka
public class KafkaConfiguration {

  @Bean
  ConcurrentKafkaListenerContainerFactory<String, Customer> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Customer> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    // less than number of partition, will do infinite retry
    factory.setConcurrency(1);
    SeekToCurrentErrorHandler errorHandler =
        new SeekToCurrentErrorHandler((record, exception) -> {
          LOGGER.info("***in error handler data, {}", record);
        }, 1);
    factory.setErrorHandler(errorHandler);
    return factory;
  }

  @Bean
  public ConsumerFactory<String, Customer> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
  }

  @Bean
  public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
    props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class);

    props.put("schema.registry.url", "http://127.0.0.1:8081");
    props.put("specific.avro.reader", "true");

    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    return props;
  }
}


@Component
@Slf4j
public class KafkaConsumerService {

  @KafkaListener(id = "demo-consumer-stream-group", topics = "kafka-demo-avro")
  public void process(ConsumerRecord<String, Customer> record) {
    LOGGER.info("Customer key: {} and value: {}", record.key(), record.value());
    LOGGER.info("topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
    throw new RuntimeException("force to retry");
  }
}

因此，如果我的侦听器中发生了异常，如果我指定的并发级别小于我的主题的分区计数，即使我在配置中配置了maxFailures，消费者也会永远重试失败的消息。

只有当我至少每隔一秒发送一条消息时，它才会起作用。如果我以批处理的方式发送消息，这种行为就不会起作用。除非我重新启动消费者，它将正常工作。

重现步骤：1.创建一个包含1个以上分区的主题，例如3或6 2。在Spring kafka 配置中，将并发级别指定为 1 3。对于“寻道到当前错误处理程序”，将最大失败指定为正值，例如 3 4。向主题发送十几条消息

您将看到每个失败的消息将进行无限重试，而不是我指定的< code>maxFailure。此外，我可以看到许多信息落后于消费者的滞后。

但是，如果您停止监听器并再次启动监听器，它将正确地跳过失败的消息。

Answer 1

我正在使用2.2.8，我仍然可以重现该错误，但配置更加复杂。我需要同时具有“寻转电流”和“重试指数随机回退策略”的模板。因此，仍然存在遇到它的风险。在这里报告。

Answer 2

这是Spring Kafka2.2.7.RELEASE的一个bug，但已在2.2.8.RELEASE中修复。