Kafka 发送重复消息

我们使用 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)]

From 关于消费者配置的 Kafka 文档 http://kafka.apache.org/documentation.html#newconsumerconfigs:

session.timeout.ms(默认 30000) - 用于检测的超时 使用 Kafka 的组管理工具时出现故障。当一个 在会话超时时间内没有收到消费者的心跳， 经纪人会将消费者标记为失败并重新平衡该组。自从 仅当以下情况时才发送心跳poll()被调用时，更高的会话 超时允许消费者有更多时间在轮询中处理消息 循环的代价是需要更长的时间来检测硬故障。也可以看看max.poll.records另一个控制处理时间的选项 轮询循环。请注意，该值必须在允许范围内 在代理配置中配置group.min.session.timeout.ms and group.max.session.timeout.ms.

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

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

另一种选择是在使用时异步处理消息pause() http://kafka-python.readthedocs.io/en/master/apidoc/kafka.consumer.html#kafka.consumer.KafkaConsumer.pause在处理消息之前和resume() http://kafka-python.readthedocs.io/en/master/apidoc/kafka.consumer.html#kafka.consumer.KafkaConsumer.resume处理消息后。在这种情况下，消费者会调用poll()（并发送心跳）即使处理时间长于session.timeout.ms。因此，经纪人不会将您的消费者标记为失败，也不会启动重新平衡。