Interval for background heartbeats unreliable

[berupp]

Hi,

This applies to 1.4.1, 1.4.3 and 1.4.4:

One of our python applications has issues with the consumer disconnecting from the broker:

[W 181210 02:02:13 base:964] Heartbeat session expired, marking coordinator dead
[W 181210 02:02:13 base:698] Marking the coordinator dead (node 1002) for group cg_0: Heartbeat session expired.

After some investigation, I ran a packet capture to verify that the heartbeat is actually sent at the configured interval (10s in our case)

The results were quite surprising. Here is the timeline for Heartbeats:
16.63
26.72
36.79
51.91
72.02
112.20
132.33
142.43
177.62
187.69
197.77
222.94
233.01
243.10
283.40

Not only are the heartbeats completely inconsistent, there are gaps of > 40 seconds.
Our session_timeout was 30 seconds, which explained the consumer disconnects.

I raised the session_timeout to 3 minutes, but still eventually missed heartbeats during a soak, leading to consumer disconnect.

I ran a simultaneous capture of two other apps written in Golang and Java. Those have 3 second heartbeats configured and where consistently spot on.

Is this something that could be fixed, or simply a limitation due to GIL?

[tvoinarovskyi]

What is your retry_backoff_ms setting? GIL should not usually have any more latency than 0.005s... What Python version are you using?

[berupp]

Python 2.7.5
Kafka 1.1.0

retry_backoff_ms is default (100)

consumer = KafkaConsumer(bootstrap_servers=self.kafka_url, group_id='cg_0',
                                     auto_offset_reset='earliest', session_timeout_ms=30000,
                                     heartbeat_interval_ms=10000, reconnect_backoff_ms=5000,
                                     reconnect_backoff_max_ms=30000)

I have since increased session_timeout to 180000 and later 300000 which is our broker maximum.

This of course does not affect the heartbeat issue, it simply mitigates it somewhat.

[xiumu-beta]

I have the same problem, I am so sad. Have you solved it ?

[berupp]

No, we simply increased the session_timeout to work around the issue for now.

I assume that the unreliable background 'thread' execution is due to python 2.7, and that upgrading the application to python 3 will fix or at least improve the condition.

I haven't had time to verify the approach. Maybe The maintainer could give his opinion on that theory.

[dpkp]

This may be related to the lock contention between senders (here, heartbeat thread) and pollers (consumer iterator perhaps). See #981 for some more details. We've landed a few PRs to master to address this. If you are able to run/test with master, give it a try. Otherwise I believe this should be improved substantially in the next release.

[jeffwidman]

@berupp was this solved by the latest 1.4.6 release?

[DieterDePaepe]

I tested the latest version (1.4.7.dev0 - commit cee4d17df7858439e0dbdf3914ca0107e080af7d), on Python 3.7. The issue still persists, resulting in frequent rebalances in our application.

The configuration:

data_consumer = kafka.KafkaConsumer(group_id=self.kafka_consumergroup,
                                            bootstrap_servers=[self.kafka_host],
                                            enable_auto_commit=False,
                                            max_poll_interval_ms=12*60_000,
                                            session_timeout_ms=3*60_000,
                                            auto_offset_reset="earliest")

The session timeout is configured for 3 minutes. (The max poll interval of 12 minutes is because during initialization of the consumer, it needs to load in data, which takes a long time. But this shouldn't be relevant.)

Here are the logs for one such rebalance, from the last successful heartbeat until the moment of failure. At this point the program was fully initialized, so each message would be processed in less than a second. We are using the iterator interface to retrieve messages.

At one point in the logs, there is simply a jump of more than 3 minutes, without any clear indication why. No heartbeats are sent in this interval, resulting in a timeout.

2019-05-27 09:15:14,747 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1194] kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58
2019-05-27 09:15:14,747 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1194, member_id='kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58')
2019-05-27 09:15:14,747 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114421: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1194, member_id='kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58')
2019-05-27 09:15:14,748 - kafka.protocol.parser - DEBUG - Received correlation id: 114421
2019-05-27 09:15:14,748 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 09:15:14,748 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114421 (0.9970664978027344 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 09:15:14,748 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 09:15:14,749 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 735847
2019-05-27 09:15:14,749 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1381491
2019-05-27 09:15:14,749 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 09:15:14,749 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:14,749 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 776: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:14,750 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 09:15:14,750 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:14,750 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114422: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:15,251 - kafka.protocol.parser - DEBUG - Received correlation id: 776
2019-05-27 09:15:15,251 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:15,251 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 776 (501.5134811401367 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=735847, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:15,252 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 735847
2019-05-27 09:15:15,252 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 09:15:15,252 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:15,252 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 777: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:15,253 - kafka.protocol.parser - DEBUG - Received correlation id: 114422
2019-05-27 09:15:15,253 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:15,253 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114422 (503.21221351623535 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1381491, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:15,754 - kafka.protocol.parser - DEBUG - Received correlation id: 777
2019-05-27 09:15:15,754 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:15,755 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 777 (501.8925666809082 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=735847, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:15,755 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 735847
2019-05-27 09:15:15,755 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1381491
2019-05-27 09:15:15,756 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 09:15:15,756 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:15,756 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 778: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:15,756 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 09:15:15,756 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:15,757 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114423: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:16,258 - kafka.protocol.parser - DEBUG - Received correlation id: 114423
2019-05-27 09:15:16,258 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:16,259 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114423 (501.7704963684082 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1381491, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:16,259 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1381491
2019-05-27 09:15:16,259 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 09:15:16,260 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:16,260 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114424: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:16,261 - kafka.protocol.parser - DEBUG - Received correlation id: 778
2019-05-27 09:15:16,261 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:16,261 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 778 (504.64463233947754 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=735847, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:16,761 - kafka.protocol.parser - DEBUG - Received correlation id: 114424
2019-05-27 09:15:16,761 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:16,761 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114424 (501.2786388397217 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1381491, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:16,762 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 735847
2019-05-27 09:15:16,762 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1381491
2019-05-27 09:15:16,762 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 09:15:16,763 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:16,763 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 779: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:16,763 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 09:15:16,763 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:16,763 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114425: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:17,265 - kafka.protocol.parser - DEBUG - Received correlation id: 114425
2019-05-27 09:15:17,265 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:17,266 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114425 (501.8887519836426 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1381491, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:17,266 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1381491
2019-05-27 09:15:17,267 - kafka.protocol.parser - DEBUG - Received correlation id: 779
2019-05-27 09:15:17,267 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:17,267 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 09:15:17,267 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 779 (504.2123794555664 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=735847, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:17,267 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:17,268 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114426: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1381491, max_bytes=1048576)])])
2019-05-27 09:15:17,269 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 735847
2019-05-27 09:15:17,269 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 09:15:17,269 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:17,270 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 780: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=735847, max_bytes=1048576)])])
2019-05-27 09:15:17,771 - kafka.protocol.parser - DEBUG - Received correlation id: 114426
2019-05-27 09:15:17,771 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:17,771 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114426 (502.96568870544434 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1381491, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:15:17,772 - kafka.protocol.parser - DEBUG - Received correlation id: 780
2019-05-27 09:15:17,772 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 09:15:17,773 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 780 (502.5663375854492 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=735847, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 09:18:55,432 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1194] kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58
2019-05-27 09:18:55,433 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1194, member_id='kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58')
2019-05-27 09:18:55,433 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 114427: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1194, member_id='kafka-python-1.4.7.dev-71d93680-c7c0-401b-a9d9-1646a199ff58')
2019-05-27 09:18:55,433 - kafka.client - DEBUG - Sending metadata request MetadataRequest_v1(topics=['x.events']) to node 0
2019-05-27 09:18:55,434 - kafka.protocol.parser - DEBUG - Sending request MetadataRequest_v1(topics=['x.events'])
2019-05-27 09:18:55,434 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 781: MetadataRequest_v1(topics=['x.events'])
2019-05-27 09:18:55,434 - kafka.coordinator - WARNING - Heartbeat session expired, marking coordinator dead
2019-05-27 09:18:55,435 - kafka.coordinator - WARNING - Marking the coordinator dead (node 2) for group MP_UnknownAD: Heartbeat session expired.
2019-05-27 09:18:55,435 - kafka.protocol.parser - DEBUG - Received correlation id: 114427
2019-05-27 09:18:55,435 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 09:18:55,435 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 114427 (2.2056102752685547 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=25)

[DieterDePaepe]

2 other cases:

2019-05-27 08:40:34,174 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1192] kafka-python-1.4.7.dev-117ec702-8620-4c18-a4b3-5b3167577a51
2019-05-27 08:40:34,174 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1192, member_id='kafka-python-1.4.7.dev-117ec702-8620-4c18-a4b3-5b3167577a51')
2019-05-27 08:40:34,175 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 99579: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1192, member_id='kafka-python-1.4.7.dev-117ec702-8620-4c18-a4b3-5b3167577a51')
2019-05-27 08:40:34,175 - kafka.protocol.parser - DEBUG - Received correlation id: 99579
2019-05-27 08:40:34,175 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 08:40:34,175 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 99579 (0.7371902465820312 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 08:40:34,176 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 08:43:38,183 - kafka.client - DEBUG - Sending metadata request MetadataRequest_v1(topics=['x.events']) to node 2
2019-05-27 08:43:38,184 - kafka.protocol.parser - DEBUG - Sending request MetadataRequest_v1(topics=['x.events'])
2019-05-27 08:43:38,184 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 99580: MetadataRequest_v1(topics=['x.events'])
2019-05-27 08:43:38,185 - kafka.coordinator - WARNING - Heartbeat session expired, marking coordinator dead
2019-05-27 08:43:38,185 - kafka.coordinator - WARNING - Marking the coordinator dead (node 2) for group MP_UnknownAD: Heartbeat session expired.
2019-05-27 08:43:38,186 - kafka.protocol.parser - DEBUG - Received correlation id: 99580
2019-05-27 08:43:38,186 - kafka.protocol.parser - DEBUG - Processing response MetadataResponse_v1

2019-05-27 07:43:37,050 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 07:43:50,548 - kafka.client - DEBUG - Sending metadata request MetadataRequest_v1(topics=['x.anomalies']) to node 1
2019-05-27 07:43:50,549 - kafka.protocol.parser - DEBUG - Sending request MetadataRequest_v1(topics=['x.anomalies'])
2019-05-27 07:43:50,549 - kafka.conn - DEBUG - <BrokerConnection node_id=1 host=gangly-tarsier-kafka-1.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.1.24', 9092)]> Request 67: MetadataRequest_v1(topics=['x.anomalies'])
2019-05-27 07:43:50,551 - kafka.protocol.parser - DEBUG - Received correlation id: 67
2019-05-27 07:43:50,551 - kafka.protocol.parser - DEBUG - Processing response MetadataResponse_v1
2019-05-27 07:43:50,551 - kafka.conn - DEBUG - <BrokerConnection node_id=1 host=gangly-tarsier-kafka-1.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.1.24', 9092)]> Response 67 (2.0258426666259766 ms): MetadataResponse_v1(brokers=[(node_id=2, host='gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka', port=9092, rack=None), (node_id=1, host='gangly-tarsier-kafka-1.gangly-tarsier-kafka-headless.kafka', port=9092, rack=None), (node_id=0, host='gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka', port=9092, rack=None)], controller_id=2, topics=[(error_code=0, topic='x.anomalies', is_internal=False, partitions=[(error_code=0, partition=2, leader=0, replicas=[0], isr=[0]), (error_code=0, partition=1, leader=2, replicas=[2], isr=[2]), (error_code=0, partition=0, leader=1, replicas=[1], isr=[1])])])
2019-05-27 07:43:50,552 - kafka.cluster - DEBUG - Updated cluster metadata to ClusterMetadata(brokers: 3, topics: 1, groups: 0)
2019-05-27 07:47:34,896 - kafka.coordinator - WARNING - Heartbeat session expired, marking coordinator dead
2019-05-27 07:47:34,896 - kafka.coordinator - WARNING - Marking the coordinator dead (node 2) for group MP_UnknownAD: Heartbeat session expired.

And some cases with irregular heartbeats not resulting in a timeout. These are taken from a one hour log for a random chosen consumer. All other heartbeats were at 3 seconds intervals.

2019-05-27 07:35:13,535 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=14) at offset 1371054
2019-05-27 07:35:13,535 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 2
2019-05-27 07:35:13,535 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1371054, max_bytes=1048576)])])
2019-05-27 07:35:13,536 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 97446: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=14, offset=1371054, max_bytes=1048576)])])
2019-05-27 07:35:14,035 - kafka.protocol.parser - DEBUG - Received correlation id: 362
2019-05-27 07:35:14,035 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 07:35:14,035 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 362 (501.7385482788086 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=729814, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 07:35:14,036 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1191] kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650
2019-05-27 07:35:14,036 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:35:14,036 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 97447: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:35:14,037 - kafka.protocol.parser - DEBUG - Received correlation id: 97446
2019-05-27 07:35:14,037 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 07:35:14,037 - kafka.protocol.parser - DEBUG - Received correlation id: 97447
2019-05-27 07:35:14,037 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 07:35:14,037 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 97446 (501.4607906341553 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1371054, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'')])])
2019-05-27 07:35:14,038 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 97447 (1.1615753173828125 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 07:35:14,038 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 07:36:35,892 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1191] kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650
2019-05-27 07:36:35,892 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:36:35,892 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 97448: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:36:35,893 - kafka.protocol.parser - DEBUG - Received correlation id: 97448
2019-05-27 07:36:35,893 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 07:36:35,894 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 97448 (1.2090206146240234 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 07:36:35,894 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD

2019-05-27 07:41:05,505 - kafka.consumer.fetcher - DEBUG - Adding fetch request for partition TopicPartition(topic='x.events', partition=15) at offset 730136
2019-05-27 07:41:05,505 - kafka.consumer.fetcher - DEBUG - Sending FetchRequest to node 0
2019-05-27 07:41:05,505 - kafka.protocol.parser - DEBUG - Sending request FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=730136, max_bytes=1048576)])])
2019-05-27 07:41:05,505 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Request 889: FetchRequest_v4(replica_id=-1, max_wait_time=500, min_bytes=1, max_bytes=52428800, isolation_level=0, topics=[(topic='x.events', partitions=[(partition=15, offset=730136, max_bytes=1048576)])])
2019-05-27 07:41:05,808 - kafka.protocol.parser - DEBUG - Received correlation id: 98143
2019-05-27 07:41:05,808 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 07:41:05,808 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 98143 (306.8230152130127 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=14, error_code=0, highwater_offset=1371648, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'\x00\x00\x00\x00\x00\x14\xed\xff\x00\x00\x01M\x00\x00\x00\xd6\x0279_N\x00\x00\x00\x00\x00\x00\x00\x00\x01j\xf8<Y\x8a\x00\x00\x01j\xf8<Y\x8a\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x00\x00\x00\x01\xb4\x04\x00\x00\x000AAAAUNIT.123456789\xf6\x03{"metri...')])])
2019-05-27 07:41:05,810 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1191] kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650
2019-05-27 07:41:05,810 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:41:05,810 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 98144: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:41:05,811 - kafka.protocol.parser - DEBUG - Received correlation id: 889
2019-05-27 07:41:05,811 - kafka.protocol.parser - DEBUG - Processing response FetchResponse_v4
2019-05-27 07:41:05,812 - kafka.conn - DEBUG - <BrokerConnection node_id=0 host=gangly-tarsier-kafka-0.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.2.40', 9092)]> Response 889 (306.17475509643555 ms): FetchResponse_v4(throttle_time_ms=0, topics=[(topics='x.events', partitions=[(partition=15, error_code=0, highwater_offset=730137, last_stable_offset=-1, aborted_transactions=NULL, message_set=b'\x00\x00\x00\x00\x00\x0b$\x18\x00\x00\x01B\x00\x00\x00\xc2\x02\x07B\xd6G\x00\x00\x00\x00\x00\x00\x00\x00\x01j\xf8<Y\x8a\x00\x00\x01j\xf8<Y\x8a\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x00\x00\x00\x01\x9e\x04\x00\x00\x000AAAAUNIT.123456789\xe0\x03{"metri...')])])
2019-05-27 07:41:05,812 - kafka.protocol.parser - DEBUG - Received correlation id: 98144
2019-05-27 07:41:05,812 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 07:41:05,813 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 98144 (2.0694732666015625 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 07:41:05,813 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 07:42:34,874 - kafka.coordinator - DEBUG - Heartbeat: MP_UnknownAD[1191] kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650
2019-05-27 07:42:34,875 - kafka.protocol.parser - DEBUG - Sending request HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:42:34,875 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Request 98145: HeartbeatRequest_v1(group='MP_UnknownAD', generation_id=1191, member_id='kafka-python-1.4.7.dev-8a788970-40ab-4a1e-8737-4c0eb6c9d650')
2019-05-27 07:42:34,876 - kafka.protocol.parser - DEBUG - Received correlation id: 98145
2019-05-27 07:42:34,876 - kafka.protocol.parser - DEBUG - Processing response HeartbeatResponse_v1
2019-05-27 07:42:34,877 - kafka.conn - DEBUG - <BrokerConnection node_id=2 host=gangly-tarsier-kafka-2.gangly-tarsier-kafka-headless.kafka:9092 <connected> [IPv4 ('10.244.3.94', 9092)]> Response 98145 (1.3849735260009766 ms): HeartbeatResponse_v1(throttle_time_ms=0, error_code=0)
2019-05-27 07:42:34,877 - kafka.coordinator - DEBUG - Received successful heartbeat response for group MP_UnknownAD
2019-05-27 07:42:34,877 - kafka.consumer.fetcher - DEBUG - Adding fetched record for partition TopicPartition(topic='x.events', partition=14) with offset 1371647 to buffered record list

[dpkp]

I made several changes to consumer and client internals that I believe should fix this. Please reopen / file new bug reports if issue persists in 1.4.7+