KafkaConsumer dies with IndexError: pop from an empty deque

Since updating to 1.4.0 this error started showing up after running the consumer for some seconds:

    for msg in self.consumer:
  File "/home/user/.local/lib/python2.7/site-packages/kafka/vendor/six.py", line 561, in next
    return type(self).__next__(self)
  File "/home/user/.local/lib/python2.7/site-packages/kafka/consumer/group.py", line 1117, in next
    return next(self._iterator)
  File "/home/user/.local/lib/python2.7/site-packages/kafka/consumer/group.py", line 1061, in _message_generator
    self._client.poll(timeout_ms=poll_ms)
  File "/home/user/.local/lib/python2.7/site-packages/kafka/client_async.py", line 577, in poll
    responses.extend(self._fire_pending_completed_requests())
  File "/home/user/.local/lib/python2.7/site-packages/kafka/client_async.py", line 669, in _fire_pending_completed_requests
    response, future = self._pending_completion.popleft()
IndexError: pop from an empty deque

On my test machine this happens always after a few seconds of running the consumer.
Sometimes the consumer dies completely sometimes it carries on.
From my debugging it seems that if it hits the heartbeat Thread, the consumer won't die but if it hits my mainthread it's game over.

[dpkp]

Thanks for the bug report -- this looks like a thread coordination issue that can happen at high loads. Unfortunately we did not catch this during pre-release testing.

Thank you for having a look at it.
This didn't happen in a high load scenario. Quite the opposite... our consumers are all rather low traffic and this happend on my test machine where it's even lower.

From my limited understanding this is what happens:
First the code in question:

    def _fire_pending_completed_requests(self):
        responses = []
        while self._pending_completion:
            thread = threading.current_thread()
            response, future = self._pending_completion.popleft()
            future.success(response)
            responses.append(response)
        return responses

Thread 1 and 2 are the main thread and the heartbeat thread. However the roles of both are interchangable.

1. Both threads hit _fire_pending_completed_requests and process self._pending_completion until one item is left
2. Now thread 1 hits the conditional in the while loop again and gets back True
3. Thread 2 gets scheduled and hits the conditional and gets back True
4. One of them continues and empties self._pending_completion
5. The other tries to popleft and triggers the exception

[dpkp]

Your analysis is correct. PR #1372 should fix. I'm surprised that our automated testing never raised this exception.

[dpkp]

This should be fixed in 1.4.1

[cheng10]

I had this issue when using 1.4.0 too. Thanks for fixing it!