SpringKafka消费者重试，回退间隔长，给出“组织.apache.kafka.clients.消费者.Commit失败异常”

Question

我是Spring-Kafka的新手，在使用Spring Kafka RetryTemplate处理kafka消息期间，尝试在失败或任何异常的情况下实现重试。

我使用了以下代码:

//这是KafkaListenerContainerFactory：

public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactoryRetry() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRetryTemplate(retryTemplate());
    factory.setRecoveryCallback(retryContext -> {
        ConsumerRecord consumerRecord = (ConsumerRecord) retryContext.getAttribute("record");
        logger.info("Recovery is called for message {} ", consumerRecord.value());
        return Optional.empty();
    });
    return factory;
}

重试模板

public RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
    // Todo: take from config
    fixedBackOffPolicy.setBackOffPeriod(240000);// 240seconds
    retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
    SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
    // Todo: take from config
    simpleRetryPolicy.setMaxAttempts(3);
    retryTemplate.setRetryPolicy(simpleRetryPolicy);
    return retryTemplate;
}
    
//

这是消费者工厂

public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

当任何异常发生时，它会按照重试策略按预期重试。一旦max重试耗尽，它就会调用恢复回调方法。但很快，它会给出“java.lang.IllegalStateException：此错误处理程序无法处理org.apache.kafka.clients.consumer.CommitFailedException；没有可用的记录信息”，其中包含一些细节，例如：失败的OffsetCommit请求，因为消费者不是活动组的一部分。

它似乎无法提交偏移量，因为消费者现在从组中启动，因为它在下一次轮询之前长时间处于空闲状态（回退期*（maxretry-1））。

我需要添加一些大价值的max.poll.interval.ms吗？

有没有其他方法可以实现这一点，这样即使消费者在处理过程中花费了很多时间并计划以长时间间隔重试，也不会出现提交失败错误。

请帮帮我。

Answer 1

聚合backOff延迟必须小于max.poll.interval。ms以避免重新平衡。

现在最好使用＜code＞SeekToCurrentErrorHandler＜/code＞而不是＜code＞RetryTemplate＜/code>，因为只有每个延迟（而不是聚合）需要小于＜code＞max.poll.interval.ms＜/code＞

此处为文档。