如果SET Spring.kafka.listener.ack mode =时间，重试的工作会吗?或在指定的ACK模式中

Question

spring kafka-如何用@kafkalistener重试

spring-kafka SeekToCurrentErrorHandler如果设置spring.kafka.listener.ack-mode=time，重试方法会工作吗？或在指定的ACK模式中重试工作，例如手动手册_immediate。

@Value("${${sync.kafka.header.source.id}}")
private String headerSourceId;
@Value("${sync.kafka.from.id}")
private String syncKafkaFromId;
@Value("${sync.kafka.to.id}")
private String syncKafkaToId;
@Autowired
private KafkaTemplate kafkaTemplate;
@Autowired
private ConsumerFactory consumerFactory;
@KafkaListener(topics = "${sync.kafka.topics}")
public void listen(ConsumerRecord<?, ?> record,
                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                   @Header(value = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) String messageKey,
                   @Header(value = "${sync.kafka.header.source.id}", required = false) String customHeader,
                   @Headers MessageHeaders messageHeaders) {
    log.info("- - - - - - - - - - - - - - -");
    log.info("topic: {}", topic);
    log.info("message key: {}", messageKey);
    log.info("custom header: {}", customHeader);
    log.info("messageHeaders: {}", messageHeaders);
    ProducerRecord<Object, Object> ProducerRecord = new ProducerRecord(topic,record.key(), record.value());
    ProducerRecord.headers().add(new RecordHeader(headerSourceId, syncKafkaFromId.getBytes()));
    kafkaTemplate.send(ProducerRecord);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
    // or factory.setRetryTemplate(aRetryTemplate);
    // and factory.setRecoveryCallback(aRecoveryCallback);
    return factory;
}

Answer 1

使用SeekToCurrentErrorHandler时，您仅应使用手册，记录ACK模式。您还应该将ackOnError容器属性设置为false，并将enable.auto.commit设置为false。