Spring kafka不重试未提交的偏移量

@KafkaListener(topics = "myTopic")
public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
    try {
        // 处理消息
        processMessage(record.value());
        acknowledgment.acknowledge(); // 手动提交偏移量
    } catch (Exception e) {
        // 处理异常
        handleException(e);
    }
}

@Bean
public SeekToCurrentErrorHandler errorHandler() {
    return new SeekToCurrentErrorHandler((record, ex) -> {
        // 自定义错误处理逻辑
        log.error("Failed to process message: {}", record.value(), ex);
    }, new FixedBackOff(1000L, 3L)); // 每秒重试一次，最多重试3次
}

@KafkaListener(topics = "myTopic", errorHandler = "errorHandler")
public void listen(ConsumerRecord<String, String> record) {
    // 处理消息
    processMessage(record.value());
}

@Transactional
@KafkaListener(topics = "myTopic")
public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
    try {
        // 处理消息
        processMessage(record.value());
        acknowledgment.acknowledge(); // 手动提交偏移量
    } catch (Exception e) {
        // 处理异常
        handleException(e);
        throw new RuntimeException("Failed to process message", e); // 抛出异常以触发事务回滚
    }
}