"socket.error: [Errno 104] Connection reset by peer" on node failure

[alexsbromberg]

I'm getting the following error from my KafkaConsumer whenever a node in my Kafka cluster goes down, even if the rest of the cluster is still running normally:

Traceback (most recent call last):
File "python/server.py", line 84, in init
File "/usr/local/lib/python2.7/site-packages/six.py", line 558, in next
return type(self).__next__(self)
File "/usr/local/lib/python2.7/site-packages/kafka/consumer/group.py", line 863, in __next__
File "/usr/local/lib/python2.7/site-packages/kafka/consumer/group.py", line 803, in _message_generator
return next(self._iterator)
self._client.poll(timeout_ms=poll_ms, sleep=True)
File "/usr/local/lib/python2.7/site-packages/kafka/client_async.py", line 439, in poll
unexpected_data = key.fileobj.recv(1)
File "/usr/local/lib/python2.7/site-packages/kafka/client_async.py", line 474, in _poll
responses.extend(self._poll(timeout, sleep=sleep))
socket.error: [Errno 104] Connection reset by peer
Exception AttributeError: "'NoneType' object has no attribute 'info'" in <bound method KafkaProducer.__del__ of <kafka.producer.kafka.KafkaProducer object at 0x7f890ff46950>> ignored

The line File "python/server.py", line 84, in init references the beginning of my consumer loop: for message in consumer

I set up my consumer like this: consumer = KafkaConsumer(*topics, group_id=group_id, bootstrap_servers=host, api_version='0.9')

Is this expected to happen? Is there something i need to do to make sure the consumer can handle nodes of the cluster failing?

Here's what I'm running on:
kafka-python 1.2.5
python 2.7
kafka api 0.9.0.1

[dpkp]

Interesting. Could be a bug in this code:

           if not conn.in_flight_requests:
                # if we got an EVENT_READ but there were no in-flight requests, one of
                # two things has happened:
                #
                # 1. The remote end closed the connection (because it died, or because
                #    a firewall timed out, or whatever)
                # 2. The protocol is out of sync.
                #
                # either way, we can no longer safely use this connection
                #
                # Do a 1-byte read to clear the READ flag, and then close the conn
                unexpected_data = key.fileobj.recv(1)
                if unexpected_data:  # anything other than a 0-byte read means protocol issues
                    log.warning('Protocol out of sync on %r, closing', conn)
                conn.close()
                continue

It looks like the call to key.fileobj.recv(1) may be raising an unhandled socket.error. Let me see if I can reproduce.

[dpkp]

Please take a look at #792 .