偏移量的异步自动提交失败 [英] Asynchronous auto-commit of offsets fails
问题描述
我有一个关于 Kafka 自动提交机制的问题.我正在使用启用了自动提交的 Spring-Kafka.作为一个实验,我在系统空闲时断开了消费者与 Kafka 的连接 30 秒(主题中没有新消息,没有消息正在处理).重新连接后,我收到了一些消息,如下所示:
I have a question on Kafka auto-commit mechanism. I'm using Spring-Kafka with auto-commit enabled. As an experiment, I disconnected my consumer's connection to Kafka for 30 seconds while the system was idle (no new messages in the topic, no messages being processed). After reconnecting I got a few messages like so:
Asynchronous auto-commit of offsets {cs-1915-2553221872080030-0=OffsetAndMetadata{offset=19, leaderEpoch=0, metadata=''}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
首先,我不明白要提交什么?系统空闲(所有先前的消息都已提交).二、断开时间为30秒,远小于5分钟(300000毫秒)max.poll.interval.ms第三,在 Kafka 失控的故障中,我收到了至少 30K 条此类消息,通过重新启动进程解决了这个问题.为什么会发生这种情况?
First, I don't understand what is there to commit? The system was idle (all previous messages were already committed). Second, the disconnection time was 30 seconds, much less than the 5 minutes (300000 ms) max.poll.interval.ms Third, in an uncontrolled failure of Kafka I got at least 30K messages of this type, which was resolved by restarting the process. Why is this happening?
我在这里列出了我的消费者配置:
I'm listing here my consumer configuration:
allow.auto.create.topics = true
auto.commit.interval.ms = 100
auto.offset.reset = latest
bootstrap.servers = [kafka1-eu.dev.com:9094, kafka2-eu.dev.com:9094, kafka3-eu.dev.com:9094]
check.crcs = true
client.dns.lookup = default
client.id =
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = feature-cs-1915-2553221872080030
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = SSL
send.buffer.bytes = 131072
session.timeout.ms = 15000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = [hidden]
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = /home/me/feature-2553221872080030.keystore
ssl.keystore.password = [hidden]
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = /home/me/feature-2553221872080030.truststore
ssl.truststore.password = [hidden]
ssl.truststore.type = JKS
value.deserializer = class org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
推荐答案
首先,我不明白要提交什么?
First, I don't understand what is there to commit?
您说得对,如果没有新数据流动,则没有任何新可提交.但是,启用 auto.commit 并且您的使用者仍在运行(即使无法连接到代理),poll 方法仍然负责以下步骤:
You are right, there is nothing new to commit if no new data is flowing. However, having auto.commit enabled and your consumer is still running (even without being able to connect to broker) the poll method is still responsible of the following steps:
- 从分配的分区获取消息
- 触发分区分配(如有必要)
- 如果启用自动偏移提交,则提交偏移
连同您的 100 毫秒间隔(参见 auto.commit.intervals
),消费者仍然尝试异步提交消费者的(未更改的)偏移位置.
Together with your interval of 100ms (see auto.commit.intervals
) the consumer still tries to asynchronously commit the (non changing) offset position of the consumer.
其次,断开时间为30秒,远小于5分钟(300000毫秒)max.poll.interval.ms
Second, the disconnection time was 30 seconds, much less than the 5 minutes (300000 ms) max.poll.interval.ms
导致重新平衡的不是 max.poll.interval 而是您的 heartbeat.interval.ms
设置和 session.timeout.ms
的组合.您的使用者根据间隔设置发送后台线程心跳,在您的情况下为 3 秒.如果在此会话超时到期(在您的情况下为 15 秒)之前代理未收到心跳,则代理将从组中删除此客户端并启动重新平衡.
It is not the max.poll.interval that is causing the rebalance but rather the combination of your heartbeat.interval.ms
setting and the session.timeout.ms
. Your consumer sends in a background thread heartbeats based on the interval setting, in your case 3 seconds. If no heartbeats are received by the broker before the expiration of this session timeout (in your case 15 seconds), then the broker will remove this client from the group and initiate a rebalance.
关于消费者配置的Kafka文档中给出了我提到的配置的更详细描述一个>
A more detailed description of the configuration I mentioned are given in the Kafka documentation on Consumer Configs
第三,在 Kafka 失控的故障中,我收到了至少 30K 条此类消息,通过重新启动进程解决了这个问题.为什么会发生这种情况?
Third, in an uncontrolled failure of Kafka I got at least 30K messages of this type, which was resolved by restarting the process. Why is this happening?
这似乎是前两个问题的组合,无法发送心跳,而消费者仍然试图通过连续调用的 poll 方法提交.
That seems to be a combination of the first two questions, where heartbeats cannot be sent and still the consumer is trying to commit through the contiuously called poll method.
正如@GaryRussell 在他的评论中提到的,我会小心使用 auto.commit.enabled
而不是自己控制偏移管理.
As @GaryRussell mentioned in his comment, I would be careful to use auto.commit.enabled
and rather take the control over the Offset Management to yourself.
这篇关于偏移量的异步自动提交失败的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!