Spring kafka批处理侦听器-批量手动提交偏移量

Question

我正在实现spring kafka批处理侦听器，它读取来自kafka主题的消息列表，并将数据发布到REST服务。我想了解在REST服务停止的情况下的偏移管理，不应该提交批处理的偏移，应该为下一次轮询处理消息。我已经阅读了spring kafka文档，但在理解侦听器错误处理程序和批量查找当前容器错误处理程序之间的区别时存在困惑。我使用的是spring-boot-2.0.0。M7及以下版本是我的代码。

Listener Config:

@Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        factory.setConcurrency(Integer.parseInt(env.getProperty("spring.kafka.listener.concurrency")));
        // factory.getContainerProperties().setPollTimeout(3000);
        factory.getContainerProperties().setBatchErrorHandler(kafkaErrorHandler());

        factory.getContainerProperties().setAckMode(AckMode.BATCH);
        factory.setBatchListener(true);
        return factory;
    }
@Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("spring.kafka.bootstrap-servers"));
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
                env.getProperty("spring.kafka.consumer.enable-auto-commit"));
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
                env.getProperty("spring.kafka.consumer.auto-commit-interval"));
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, env.getProperty("spring.kafka.session.timeout"));
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("spring.kafka.consumer.group-id"));
        return propsMap;
    }

Listener Class:

@KafkaListener(topics = "${spring.kafka.consumer.topic}", containerFactory = "kafkaListenerContainerFactory")
    public void listen(List<String> payloadList) throws Exception {
        if (payloadList.size() > 0)
            //Post to the service
    }

Kafka Error Handler:

public class KafkaErrorHandler implements BatchErrorHandler {

    private static Logger LOGGER = LoggerFactory.getLogger(KafkaErrorHandler.class);

    @Override
    public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
        LOGGER.info("Exception occured while processing::" + thrownException.getMessage());

            }

}

如何处理Kafka listener，以便在处理一批记录时发生问题，我不会丢失数据。

Answer 1

使用Apache Kafka，我们永远不会丢失数据。分区日志中确实存在一个偏移量，可以搜索到任意位置。

另一方面，当我们使用分区中的记录时，不需要提交它们的偏移量——当前的使用者将状态保存在内存中。我们只需要为同一群体中的其他新消费者做出promise，而当前消费者已经死亡。与错误无关，当前使用者总是继续轮询其当前内存偏移量后面的新数据。

因此，要在同一消费者中重新处理相同的数据，我们必须使用seek操作将消费者移回所需位置。这就是为什么Spring Kafka引入了seekToCurInterrorHandler：

这允许实现查找所有未处理的主题/分区，以便在下次轮询时检索当前记录（以及剩余的其他记录）。seekToCurInterrorHandler就是这样做的。

https://docs.spring.io/spring-kafka/reference/htmlsingle/#_seek_to_current_container_error_handlers