SpringKafka-偏移量未设置为on错误

Question

在我的侦听器中，在使用消息后，如果发生任何异常，我将抛出一个异常。如果它成功了，那么我承认。但是，即使抛出异常，偏移量也不会后退。i、 e重试没有按预期进行。错误事件不会再次出现。

此外，我发现我没有消费所有预期的消息。我做错什么了吗？

@Bean
public ConcurrentKafkaListenerContainerFactory<String,Result<FormatException,String>> kafkaListenerContainerFactory(){
    ConcurrentKafkaListenerContainerFactory<String, Result<FormatException, String>> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(this.consumerFactory());
    factory.setErrorHandler(new SeekToCurrentErrorHandler(3));
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

    return factory;
}


public ConsumerFactory<String, Result<FormatException, String>> consumerFactory() {
   kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    kafkaProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
}

监听器类

@KafkaListener(topics = "${gpc.consumer.topic}")
    public void listenWithHeadersGPC(ConsumerRecord<String, Result<FormatException, String>> consumerRecord,
            Acknowledgment acknowledgment) {
        try {
            String body = consumerRecord.value().get();
            log.logInfo("Received message for " + GPCSourceName + ": " + body);
            log.logInfo("consumption successful from kafka. partition="+consumerRecord.partition()+", "
                    + "offset="+consumerRecord.offset());
                            if(body.contains("[e-mail]")) {
                                throw new RuntimeException("custom exception");
                            }
                            log.logInfo("Filtered message for " + GPCSourceName + " : " + body);
            acknowledgment.acknowledge();
        }catch(Exception e) {
            log.logError("Exception occurred in listenWithHeadersGPC: ", e);
            throw e;
        }
    }

我正在做确认。承认（） 在我的侦听器中，如果没有异常

编辑：SpringKafka版本：2.2.8




Answer 1

工厂。setErrorHandler（新的SeekTocurErrorHandler（3））

在这种配置下，你会看到3次重试，两次之间的延迟为零；最后记录下来。默认情况下，“已恢复”记录的偏移量不会提交。

发布这样的问题时；指出您使用的版本。

从2.3版开始，您可以使用确认。nack（睡眠时间）；您还可以在seekTocurInterrorHandler上调用setCommitRecovered（true），恢复记录的偏移量将被提交（使用手动\u IMMEDIATE）。

只有当异常被抛出到容器时，这一切才会发生；如果在侦听器中捕捉到异常，则不会看到重试。

如果您根本看不到重试，并且您的侦听器正在抛出异常，那么您的侦听器或配置有问题。显示所有配置和@KafkaListener代码。