2

我们使用 kafka(0.9.0.0) 来编排不同微服务之间的命令消息。我们发现了一个间歇性问题，其中重复消息被传递到特定主题。下面给出了发生此问题时发生的日志。有人可以帮助理解这个问题吗 

Wed, 21-Sep-2016 09:19:07 - WARNING Coordinator unknown during heartbeat -- will retry
Wed, 21-Sep-2016 09:19:07 - WARNING Heartbeat failed; retrying
Wed, 21-Sep-2016 09:19:07 - WARNING <BrokerConnection host=AZSG-D-BOT-DEV4 port=9092> timed out after 40000 ms. Closing connection.
Wed, 21-Sep-2016 09:19:07 - ERROR Fetch to node 1 failed: RequestTimedOutError - 7 - This error is thrown if the request exceeds the user-specified time limit in the request.
Wed, 21-Sep-2016 09:19:07 - INFO Marking the coordinator dead (node 1): None.
Wed, 21-Sep-2016 09:19:07 - INFO Group coordinator for kafka-python-default-group is BrokerMetadata(nodeId=1, host=u'AZSG-D-BOT-DEV4', port=9092)
Wed, 21-Sep-2016 09:19:07 - ERROR OffsetCommit failed for group kafka-python-default-group due to group error (UnknownMemberIdError - 25 - Returned from group requests (offset commits/fetches, heartbeats, etc) when the memberId is not in the current generation.), will rejoin
Wed, 21-Sep-2016 09:19:07 - WARNING Offset commit failed: group membership out of date This is likely to cause duplicate message delivery.
Wed, 21-Sep-2016 09:19:07 - ERROR LeaveGroup request failed: UnknownMemberIdError - 25 - Returned from group requests (offset commits/fetches, heartbeats, etc) when the memberId is not in the current generation.
Wed, 21-Sep-2016 09:19:07 - INFO Marking the coordinator dead (node 1): None.
Wed, 21-Sep-2016 09:19:07 - INFO Group coordinator for kafka-python-default-group is BrokerMetadata(nodeId=1, host=u'AZSG-D-BOT-DEV4', port=9092)
Wed, 21-Sep-2016 09:19:07 - ERROR OffsetCommit failed for group kafka-python-default-group due to group error (UnknownMemberIdError - 25 - Returned from group requests (offset commits/fetches, heartbeats, etc) when the memberId is not in the current generation.), will rejoin
Wed, 21-Sep-2016 09:19:07 - WARNING Offset commit failed: group membership out of date This is likely to cause duplicate message delivery.
Wed, 21-Sep-2016 09:19:10 - INFO Joined group 'kafka-python-default-group' (generation 5) with member_id kafka-python-1.0.2-8585f310-cb4f-493a-a98d-12ec9810419b
Wed, 21-Sep-2016 09:19:10 - INFO Updated partition assignment: [TopicPartition(topic=u'ilinaTestPlatformReq', partition=0)]
4

1 回答 1

6

来自消费者配置的 Kafka 文档：

session.timeout.ms(默认 30000) - 使用 Kafka 的组管理工具时用于检测故障的超时时间。当会话超时内没有收到消费者的心跳时，代理会将消费者标记为失败并重新平衡组。由于只有在poll()被调用时才会发送心跳，因此较高的会话超时允许有更多时间在消费者的轮询循环中进行消息处理，但代价是检测硬故障的时间更长。另请参阅 max.poll.records另一个选项来控制轮询循环中的处理时间。请注意，该值必须在代理配置中由group.min.session.timeout.ms 和配置的允许范围内group.max.session.timeout.ms。

似乎如果消息处理时间大于 30000 毫秒，就会触发消费者再平衡，这可能会导致重复消息传递。

你可以尝试的是增加session.timeout.ms。

另一种选择是异步处理消息，同时在处理消息之前使用pause()并在处理消息之后使用resume()。在这种情况下，poll()即使处理时间比session.timeout.ms. 因此经纪人不会将您的消费者标记为失败，并且不会启动重新平衡。

于 2016-09-22T09:50:52.103 回答