我有一个Kafka消费者,它定义了以下属性:
session.timeout.ms = 60000
heartbeat.interval.ms = 6000
我们注意到大约有2000个消息的延迟,并且通过我们的应用程序日志发现同一条消息被消费者多次消费。此外,注意到一些消息需要大约10秒才能完全处理。我们怀疑消费者没有正确提交偏移量(或重复提交相同的旧偏移量),因此消费者多次获取了相同的消息。
为了解决这个问题,我们引入了几个新属性:
auto.commit.interval.ms=20000 //To ensure that commit is happening only after processing of message is completed
max.poll.records=10 //To make the consumer pick only 10 messages in one go
And, we set the concurrency to 1.
这解决了我们的问题。延迟开始减少,最终降为0。
但是,我仍然不清楚为什么一开始会出现这个问题。 据我了解,默认情况下:
enable.auto.commit = true
auto.commit.interval.ms=5000
理想情况下,消费者应该每隔5秒提交一次。如果消息在这个时间内没有完全处理,会发生什么?消费者提交的偏移量是什么?问题是否由于大型轮询记录大小(默认为500)而发生?
另外,关于轮询方法poll(),我读到:
"调用poll()会在后台以设置的auto.commit.interval.ms间隔自动提交偏移量"
因此,如果最初的poll()每5秒钟就会发生一次(默认的auto.commit.interval),为什么它没有提交最新的偏移量?因为消费者还没有完成处理吗?那么,它应该在下一次5秒钟时提交该偏移量。
可以有人回答这些问题并解释原始问题为什么会出现吗?
poll()
不会在后台调用。自KIP-62以来,只有心跳包会在后台发送。当使用自动提交(不建议与Spring一起使用)时,如果时间间隔已过,则提交将仅在下一次调用poll()
时发生;如果时间间隔尚未过去,则提交将等到后续的轮询。对于Spring,使用BATCH(默认)或RECORDAckMode
意味着提交将更具确定性,与时间无关。我们现在默认将enable.auto.commit
设置为false。 - Gary Russell