我正在尝试使用Spring-Kafka(1.1.0.RELEASE)手动提交Kafka消费者的偏移量。 我明白,为了实现强大的客户端,最好提交这些偏移量,以便其他消费者不会处理可能由已死的消费者或因触发重新平衡而最初处理的重复事件。
此外,我正在使用
我看到有两种处理方式 -
Set the ACK_MODE to MANUAL_IMMEDIATE and in the listener implementation, invoke ack.acknowledge() API
ConcurrentKafkaListenerContainerFactory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE); @KafkaListener(topics = "anotherBatchTopic", containerFactory = "batchContainerFactory") public void listen(List<ConsumerRecord<byte[], String>> batch, Acknowledgment acknowledgment) throws Exception { logger.info("===*** batch size : " + batch.size() + "***==="); batch.forEach(System.out::println); acknowledgment.acknowledge(); }
Implement a ConsumerRebalanceListener.
ConcurrentKafkaListenerContainerFactory.getContainerProperties().setConsumerRebalanceListener(new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(final Collection<TopicPartition> collection) { } @Override public void onPartitionsAssigned(final Collection<TopicPartition> collection) { } });
consumer.commitSync()
或consumer.commitASync()
API。由于某些技术限制,我无法升级到支持ConsumerAwareRebalanceListener
并具有对Consumer
的引用的最新版本的spring-kafka。
那么,利用ConsumerRebalanceListener
来能够提交偏移量到Kafka的方法是什么?此外,我正在使用
ConcurrentKafkaListenerContainerFactory.setConcurrency()
API启动多个消费者监听线程,因此如果特定消费者线程失败,它是否具有自己的ConsumerRebalanceListener
实例?