Kafka enable.auto.commit 设置为 false 但轮询仍获取“下一条”消息

Question

我想在我的消费者成功处理记录时告诉 Kafka，所以我通过将 enable.auto.commit 设置为 false 来关闭自动提交。我在偏移量 0 和 1 处订阅了关于我订阅的主题的两条消息，并创建了一个使用者，以便每次调用 poll 最多返回 一个 记录（通过将 max.poll.records 设置为 1 )。

我现在拨打consumer.poll(5000) 并收到第一条消息，但我不承认；我不打电话给commitSync 或commitAsync。如果我现在再次调用consumer.poll(5000)，使用相同的消费者，我希望收到与我刚刚阅读的完全相同的消息，但相反，我收到了第二条消息。

如何让consumer.poll 继续发送相同的消息，直到我明确确认？

Answer 1

您所描述的是预期的行为。每次调用poll()，它都会返回下一条消息。您提交的偏移量仅在连接新消费者时使用，因此它知道从哪里（重新）开始。

在 MessageHub 中，我们将 session.timeout 设置为 30 秒。所以你需要稍微快一点地调用poll()，以避免被断开连接。如果您的处理时间比这更长，那么我可以考虑 2 个选项：

• 使用 Kafka 0.10.2 并设置 max.poll.interval.ms 来告诉您的 Kafka 客户端在处理之前的记录时保持会话活动（无需调用 poll()）。 （此功能是在 0.10.1 中添加的，但我们不支持该版本。0.10.2 有效，因为它能够与 0.10.0 代理一起使用）

• 使用 seek() 移动到 poll 之后的上一个偏移量，这样它就会一直返回相同的记录。

希望这会有所帮助！