Q:

Heartbeat session expired, marking coordinator dead

When I am using python-kafka(1.4.4) library to consume the message(Kafka version 1.1.0,Python 3.7).It throw this error again and again.And I don't know where going wrong,here is my Python code,the consumer initial:
consumer = KafkaConsumer('dolphin-spider-google-book-bookinfo',
                         bootstrap_servers=['mq-server:9092'],
                         group_id = "google-book",
                         client_id = "dolphin-pipline-google-bookinfo-consumer-foolman",
                         # Manage kafka offsets manual
                         enable_auto_commit = False,
                         consumer_timeout_ms=50000,
                         # consume from beginning
                         auto_offset_reset = "earliest",
                         max_poll_interval_ms =350000,
                         session_timeout_ms = 60000,
                         request_timeout_ms = 700000
                         ) 

Here is the consume logic:
def consume_bookinfo(self):
        while True:
            try:
                for books in self.consumer:
                    logger.info("Get books info offset: %s" ,books.offset)                    
                    self.sub_process_handle(books.value,books.offset)                    
            except Exception as e:
                logger.error(e)

    def sub_process_handle(self,bookinfo,offset):     
        number_of_threadings = len(threading.enumerate())
        if(number_of_threadings < 13):
            t = threading.Thread(target=self.background_process,name="offset-" + str(offset), args=(bookinfo,), kwargs={})
            t.start()
        else:
            # If all threading running
            # Using main thread to handle
            # Slow down kafka consume speed
            logger.info("Reach max handle thread,sleep 20s to wait thread release...")
            time.sleep(20)            
            self.sub_process_handle(bookinfo,offset)

    def background_process(self,bookinfo):        
        self.parse_bookinfo(bookinfo)
        self.consumer.commit_async(callback=self.offset_commit_result) 

I start multi thread to handle the consume logic.But run some time,throw this error:
2019-01-30 02:46:52,948 - /home/dolphin/source/dolphin-pipline/dolphin/biz/spider_bookinfo_consumer.py[line:37] - INFO: Get books info offset: 9304
2019-01-30 02:46:52,948 - /home/dolphin/source/dolphin-pipline/dolphin/biz/spider_bookinfo_consumer.py[line:51] - INFO: Reach max handle thread,sleep 20s to wait thread release...
2019-01-30 02:47:12,968 - /home/dolphin/source/dolphin-pipline/dolphin/biz/spider_bookinfo_consumer.py[line:61] - INFO: commit offset success,offsets: {TopicPartition(topic='dolphin-spider-google-book-bookinfo', partition=0): OffsetAndMetadata(offset=9305, metadata='')}
2019-01-30 04:27:47,322 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:964] - WARNING: Heartbeat session expired, marking coordinator dead
2019-01-30 04:27:47,323 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:698] - WARNING: Marking the coordinator dead (node 0) for group google-book: Heartbeat session expired.
2019-01-30 04:27:47,433 - /usr/local/lib/python3.5/site-packages/kafka/cluster.py[line:353] - INFO: Group coordinator for google-book is BrokerMetadata(nodeId=0, host='35.229.69.193', port=9092, rack=None)
2019-01-30 04:27:47,433 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:676] - INFO: Discovered coordinator 0 for group google-book
2019-01-30 04:27:47,433 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/consumer.py[line:341] - INFO: Revoking previously assigned partitions {TopicPartition(topic='dolphin-spider-google-book-bookinfo', partition=0)} for group google-book
2019-01-30 04:27:47,433 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:434] - INFO: (Re-)joining group google-book
2019-01-30 04:27:47,437 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:504] - INFO: Elected group leader -- performing partition assignments using range
2019-01-30 04:27:47,439 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/base.py[line:333] - INFO: Successfully joined group google-book with generation 470
2019-01-30 04:27:47,439 - /usr/local/lib/python3.5/site-packages/kafka/consumer/subscription_state.py[line:257] - INFO: Updated partition assignment: [TopicPartition(topic='dolphin-spider-google-book-bookinfo', partition=0)]
2019-01-30 04:27:47,439 - /usr/local/lib/python3.5/site-packages/kafka/coordinator/consumer.py[line:238] - INFO: Setting newly assigned partitions {TopicPartition(topic='dolphin-spider-google-book-bookinfo', partition=0)} for group google-book
2019-01-30 04:27:47,694 - /home/dolphin/source/dolphin-pipline/dolphin/biz/spider_bookinfo_consumer.py[line:63] - ERROR: commit offset failed,detail: CommitFailedError: Commit cannot be completed since the group has already
            rebalanced and assigned the partitions to another member.
            This means that the time between subsequent calls to poll()
            was longer than the configured max_poll_interval_ms, which
            typically implies that the poll loop is spending too much
            time message processing. You can address this either by
            increasing the rebalance timeout with max_poll_interval_ms,
            or by reducing the maximum size of batches returned in poll()
            with max_poll_records.

2019-01-30 04:27:47,694 - /home/dolphin/source/dolphin-pipline/dolphin/biz/spider_bookinfo_consumer.py[line:63] - ERROR: commit offset failed,detail: CommitFailedError: Commit cannot be completed since the group has already
            rebalanced and assigned the partitions to another member.
            This means that the time between subsequent calls to poll()
            was longer than the configured max_poll_interval_ms, which
            typically implies that the poll loop is spending too much
            time message processing. You can address this either by
            increasing the rebalance timeout with max_poll_interval_ms,
            or by reducing the maximum size of batches returned in poll()
            with max_poll_records.

How to avoid this problem?What should I do?

A:

Lets first look at what might be the cause of this error. As it is discussed in the official kafka consumer documentation(here) kafka detects connected consumers when they call poll()

After subscribing to a set of topics, the consumer will automatically join the group when poll(Duration) is invoked. The poll API is designed to ensure consumer liveness. As long as you continue to call poll, the consumer will stay in the group and continue to receive messages from the partitions it was assigned. Underneath the covers, the consumer sends periodic heartbeats to the server. If the consumer crashes or is unable to send heartbeats for a duration of session.timeout.ms, then the consumer will be considered dead and its partitions will be reassigned. 

So to stay in the group, you must continue to call poll. The max.poll.interval.ms setting declares how long a consumer can stay in a group without calling poll().
Each call to poll() returns many records(default is 500) which is then iterated through in the for message in consumer. Next call to poll() is made when you finished processing all of returned records.
If it takes too long for your program to process the records max.poll.interval.ms is going to pass and you will get kicked out of the group.
Here is what you can do:

Increase max.poll.interval.ms
Decrease max.poll.records

In @Dolphin answer,he is actually is decreasing max.poll.records to 1. I prefer doing it this way:
self.consumer = kafka.KafkaConsumer(topic, bootstrap_servers='servers:ports',
                       group_id='group_id',max_poll_records=1,max_poll_interval_ms=300000)

the important part is max_poll_records=1. Of course, you might want to set it to a lager value than 1.

