为什么我所有的Kafkapromise不是真的promise？(0.9.0.1)

Question

Reader 1 -> Partition 0 & 1
Reader 2 -> Partition 2 & 3

我从分区中读取，并对我读取的每个ConsumerRecord执行consumer.commitasync（在理解行为之前，在这一点上故意不对提交进行批处理）。

我在commit async回调中放置了一个每个主题的计数器来度量它被调用的次数，总数为100万次。

在应用程序稳定下来并停止之后，我使用Kafka CLI工具查看我的偏移量，我得到如下内容：

Group           Topic                          Pid Offset          logSize         Lag             Owner
group1          lowercaseStrings               0   233788          250000          16212           none
group1          lowercaseStrings               1   249999          250000          1               none
group1          lowercaseStrings               2   249999          250000          1               none
group1          lowercaseStrings               3   233788          250000          16212           none

为什么我还会有这种滞后？什么可能导致这一切？

Answer 1

Kafka批量获取消息。当您在消息流中调用next（）时，可能会发生两件事：-要么本地没有数据，用户将在内部调用poll()，这将更新其偏移量状态-要么有一些数据，您将在本地批处理中前进。

当您调用commitAsync（）时，您提交了在上次poll（）调用中获得的偏移量，而不是当前通过内部迭代有效地执行的偏移量。

您可以通过减小接受的批处理的大小（在配置中设置batch.size）来控制这种行为，如果您真的想对每个消息进行poll（），则将其降至0。我希望将此配置降至0，所有地方都将达到0滞后（注意，对于可伸缩的使用，这将破坏吞吐量）。