使用spring kafka在kafka消费者中手动提交偏移量的方法

Question

我正在尝试找出使用Spring-Kafka（1.1.0. RELEASE）在Kafka消费者中手动提交偏移的方法。我明白，最好将这些偏移提交给健壮的客户端实现，这样其他消费者就不会处理重复的事件，这些事件最初可能是由现已死亡的消费者处理的，或者因为重新平衡被触发了。

我知道有两种方法可以解决这个问题-

>

将ACK_MODE设置为MANUAL_IMMEDIATE，并在侦听器实现中调用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();
}

实现消费者重新平衡监听器。

ConcurrentKafkaListenerContainerFactory.getContainerProperties().setConsumerRebalanceListener(new ConsumerRebalanceListener() {

@Override
public void onPartitionsRevoked(final Collection<TopicPartition> collection) {

}

@Override
public void onPartitionsAssigned(final Collection<TopicPartition> collection) {

}
});

然而，使用这种方法，我不知道如何获得消费者的引用，以调用消费者。commitSync（）或使用者。commitASync（）API。由于一些技术限制，我无法使用最新版本的spring kafka，该版本支持ConsumerAwareRebalanceListener，并引用了Consumer。

那么，如何利用ConsumerBalanceListener向Kafka提交偏移量呢？

此外，我正在使用ConcurrentKafkAlisterContainerFactory启动多个消费者侦听器线程。setConcurrency（）API，那么如果特定的使用者线程死亡，它是否有自己的ConsumerBalanceListener实例？

Answer 1

消费者重新平衡侦听器不能用于提交偏移量；使用1. x；唯一的方法是通过确认参数。

1.1.0非常古老；全部1。建议x用户至少升级到1.3.5；得益于KIP-62，它的线程模型简单多了——请参见项目页面。

当前版本2.1.6（2.1.7将于今天发布）有更多选项，包括ConsumerRawareMessageListener，您可以在其中完全访问消费者。

您不能从带有版本的侦听器直接与消费者交互