问题
我们使用 StatefulSet 在 Kubernetes 上部署 Scala Kafka Streams 应用程序。实例有单独的 applicationId
s,因此它们每个都复制完整的输入主题以进行容错。它们本质上是只读服务,只读取状态主题并将其写入状态存储,从那里通过 REST 为客户请求提供服务。这意味着,在任何给定时间,消费者组始终只包含一个 Kafka Streams 实例。
我们现在的问题是,在触发滚动重启时,每个实例大约需要 5 分钟才能启动,其中大部分时间都花在了 REBALANCING
中等待。状态。我已阅读 here Kafka Streams 不发送 LeaveGroup
请求以便在容器重新启动后快速返回,而无需重新平衡。为什么这对我们不起作用,为什么重新平衡需要这么长时间,即使 applicationId
是相同的吗?理想情况下,为了最大限度地减少停机时间,应用程序应立即从重新启动时离开的地方接管。
配置
以下是我们从默认值更改的一些配置:
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "1000")
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), "300000")
properties.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest")
// RocksDB config, see https://docs.confluent.io/current/streams/developer-guide/memory-mgmt.html
properties.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, classOf[BoundedMemoryRocksDBConfig])
问题/相关配置
session.timeout.ms
?我们将其设置为相当大的值,因为 Kafka 代理位于不同的数据中心,并且网络连接有时不是非常可靠。 max.poll.interval.ms
,因为它与重新平衡超时有关。那是对的吗?我对改变这一点犹豫不决,因为它可能会对我们应用程序的正常运行模式产生影响。 group.initial.rebalance.delay.ms
在部署期间延迟重新平衡 - 但这也会在从崩溃中恢复后导致延迟,不是吗? group.instance.id
完全消除静态成员(member)资格的消费者重新平衡。 ,这将非常适合我们的用户案例,但它似乎还没有在我们的经纪人上可用。 我对众多配置以及如何使用它们在更新后启用快速恢复感到困惑。有人可以解释一下他们是如何一起玩的吗?
最佳答案
您引用的另一个问题并不是说在重新启动时避免重新平衡。不发送 LeaveGroupRequest
仅在您停止应用程序时避免重新平衡。因此,重新平衡的次数从两个减少到一个。当然,通过你有点不寻常的单实例部署,你在这里没有任何收获(事实上,它实际上可能“伤害”了你......)
Would it help to decrease session.timeout.ms? We set it to quite a large value as the Kafka brokers live in a different data center and network connections are at times not super reliable.
可能是,取决于您重新启动应用程序的速度。 (更多详细信息如下。)也许只是尝试一下(即,将其设置为 3 分钟以保持较高的稳定性值,并看到重新平衡时间下降到 3 分钟?
This answer suggests to decrease max.poll.interval.ms, as it is tied to a rebalance timeout. Is that correct? I'm hesitant to change this, as it might have consequences on the normal operation mode of our app.
max.poll.interval.ms
也会影响重新平衡时间(更多详情见下文)。但是,默认值为 30 秒,因此不应导致 5 分钟的重新平衡时间。There is mention of a config group.initial.rebalance.delay.ms to delay rebalancing during a deployment - but that would cause delays also after recovery from a crash, wouldn't it?
这仅适用于空消费者组,默认值仅为 3 秒。所以它应该不会影响你。
I also stumbled upon KIP-345, which targets to eliminate consumer rebalancing for static memberships entirely via group.instance.id, which would be a good fit for our user case, but it does not seem to be available yet on our brokers.
使用静态组成员资格实际上可能是最好的选择。也许值得升级您的经纪人以获得此功能。
顺便说一句,
session.timeout.ms
之间的区别和 max.poll.interval.ms
在另一个问题中解释:Difference between session.timeout.ms and max.poll.interval.ms for Kafka 0.10.0.0 and later versions通常,代理端组协调器维护每个“组生成”的所有成员的列表。如果成员主动离开组(通过发送
LeaveGroupRequest
)、超时(通过 session.timeout.ms
或 max.poll.interval.ms
)或新成员加入组,则会触发重新平衡。如果发生重新平衡,每个成员都有机会重新加入该组以包含在下一代中。对于您的情况,该组只有一名成员。当你停止你的应用程序时,没有
LeaveGroupRequest
发送,因此组协调员将仅在 session.timeout.ms
之后删除该成员通过。如果您重新启动应用程序,它会作为"new"成员返回(从组协调员的角度来看)。这将触发重新平衡,使该组的所有成员都可以重新加入该组。对于您的情况,“旧”实例可能仍在组中,因此只有在组协调员从组中删除旧成员后,重新平衡才会向前推进。问题可能是,组协调员认为组从一个成员扩展到两个成员......(这就是我上面的意思:如果要发送
LeaveGroupRequest
,当你阻止你时,组会变成空的应用程序,并且在重新启动时只有新成员会在组中,并且重新平衡将立即进行。)使用静态组成员身份可以避免这个问题,因为在重新启动时,实例可以被重新识别为“旧”实例,并且组协调器不需要等待旧组成员过期。
关于java - Kubernetes 上的 Kafka 流 : Long rebalancing after redeployment,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/61679271/