列出当前组配置并更新配置

Question

我们正在使用kafka来实现一个驱动事件应用程序，并在组内的消费者之间进行大量的再平衡。

CommitFailedException提交无法完成，因为组已重新平衡并将分区分配给另一个成员

我想尝试一下，但我不知道如何在代理中配置group.max.session.timeout.ms。

我还发现：

Answer 1

我不确定您启动代理和客户端的方式，但希望您使用Kafka的bin文件夹中的脚本通过控制台启动em，以下是您应该遵循的步骤：

• 代理端

例如，您应该将group.max.session.timeout.ms参数增加到double（？？）。如最新版本的文档所述，此参数定义了注册使用者允许的最大会话超时。更长的超时使使用者有更多的时间处理心跳之间的消息，代价是检测故障的时间更长。当您启动Kafka代理时，最简单的命令如下：

bin/kafka-server-start.sh config/server.properties

您必须通过添加参数来修改config/server.properties文件。仅供示例使用（更好地定位在更低的位置..）：

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker. 
broker.id=0

###group session timeout! yep, this one
group.max.session.timeout.ms=3600000  //(default is 1800000)

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from

...

重新启动Kafaka，你就完成了经纪人的任务。

• 消费者端

bin/kafka-console-consumer.sh --consumer.config config/myconsumer.properties

(...)
##Consumer session timeout! 
session.timeout.ms=20000 //(default is 10000)
(...)

也许偏离了主题，但在Kafka的配置调优中，您还可以检查是否有可能将消费和处理分离（例如，通过使用某种资源轮询）。

希望有帮助！