我正在运行一个数据管道,以便 我通过jdbc connect将sql db中的数据读入kafka主题 我使用 ES 的 kafka 接收器连接器将这些数据接收到 Elasticsearch 中
我需要重置这条管道。为此,我想重置正在侦听 jdbc 连接主题的消费者组。所以我运行以下命令
kafka-consumer-groups --reset-offsets --to-earliest --all-topics --execute --group mygroup --bootstrap-server myserver:9092
但是我收到这个错误
Error: Assignments can only be reset if the group 'mygroup' is inactive, but the current state is Stable.
我停止了连接以使该组处于非活动状态,但这不起作用。所以我的问题是,如何让我的小组不活跃
--reset-offsets
需要在重置偏移量之前检查消费者当前是否处于活动状态。 Stable
表示消费者当前处于活动状态,因此偏移量无法回移。
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group mygroup --describe
应提供任何活跃的消费者群体(如果有)。输出将帮助您识别哪些应用程序当前处于活动状态。一旦让消费者保持活跃的应用程序被终止,您将能够重置偏移量。
或者,如果您使用的是旧消费者,您可以通过运行
来删除消费者组bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --group mygroup --delete
最后重置偏移量。
我刚刚用 kafka 2.12 尝试过,当我遇到同样的错误时,我停止了持有该组的微服务(从该消费者组中的主题读取消息),这立即解决了问题。之后,我可以重新调整微服务的规模,一切都会继续。
正如其他答案解释它是如何工作的,这是因为有一个消费者附加到该主题,所以最简单的事情就是让应用程序处于非活动状态一段时间。
您也可以删除消费者组,但最终可能会遇到此处
描述的问题您可以将 Pod 或服务缩减至零或直接关闭它们。 有时您的 Pod 会自动重新启动,但这给您足够的时间来执行命令。