Async producer thread dies with KafkaUnavailableError

[mutability]

After a network outage while there were pending async produce requests, this happened:

20150719 05:12:59.928  ERROR    kafka.client         Error sending request [98903] to server aggee:9092, trying next server
Traceback (most recent call last):
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/conn.py", line 97, in _read_bytes
    data = self._sock.recv(min(bytes_left, 4096))
TimeoutError: [Errno 60] Operation timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 121, in _send_broker_unaware_request
    response = conn.recv(requestId)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/conn.py", line 155, in recv
    resp = self._read_bytes(4)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/conn.py", line 107, in _read_bytes
    self._raise_connection_error()
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/conn.py", line 82, in _raise_connection_error
    raise ConnectionError("Kafka @ {0}:{1} went away".format(self.host, self.port))
kafka.common.ConnectionError: Kafka @ aggee:9092 went away
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/local/lib/python3.4/threading.py", line 920, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.4/threading.py", line 868, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/producer/base.py", line 154, in _send_upstream
    fail_on_error=False)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 476, in send_produce_request
    resps = self._send_broker_aware_request(payloads, encoder, decoder)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 166, in _send_broker_aware_request
    payload.partition)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 86, in _get_leader_for_partition
    self.load_metadata_for_topics(topic)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 364, in load_metadata_for_topics
    resp = self.send_metadata_request(topics)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 430, in send_metadata_request
    return self._send_broker_unaware_request(payloads, encoder, decoder)
  File "/usr/home/obj/.local/lib/python3.4/site-packages/kafka/client.py", line 130, in _send_broker_unaware_request
    raise KafkaUnavailableError('All servers failed to process request')
kafka.common.KafkaUnavailableError: All servers failed to process request

After this, the async producer was stuck with a permamently full queue and didn't recover after the broker came back up.

[thieman]

This is a serious availability concern, and it looks like there's a PR to fix it that passed tests (with the exception of one Python 3.4 test which looks like a flap rather than an actual error). Can we get a review on this? @dpkp

[dpkp]

merged!