Help: Fetch to node failed because of ConnectionError

[secfree]

Exception log:

ERROR: 2016-08-17 15:55:42,092: future.py:79 Fetch to node 1001 failed: ConnectionError: <BrokerConnection host=hostname_01/ip_address_01 port=9092>

Environment:

1. Kafka version: 0.9
2. kafka-python version: 1.3.1
3. Topic has 8 partition on 8 brokers
4. Consume by KafkaConsumer with group_id

More Descriptions:

1. It's ok if the offset lag > 0 on each broker
2. It's ok if the offset lag = 0 when consumer connecting
3. There are exception log like above for 7 (which is 8 - 1) brokers if:
   1. Begin with offset lag > 0
   2. End with offset lag = 0

[dpkp]

Can you enable debug logging and post those? ConnectionError generally indicates you have a networking issue.

[secfree]

I have enabled debug logging when I am trying to fix the problem.

Here is the log:

INFO: 2016-08-17 16:50:02,614: conn.py:835 Broker is not v(0, 10) -- it did not recognize ApiVersionRequest_v0
INFO: 2016-08-17 16:50:02,778: conn.py:808 Broker version identifed as 0.9
INFO: 2016-08-17 16:50:02,778: conn.py:810 Set configuration api_version=(0, 9) to skip auto check_version requests on startup
INFO: 2016-08-17 16:50:02,780: subscription_state.py:140 Updating subscribed topics to: ('topic-test',)
INFO: 2016-08-17 16:50:02,783: cluster.py:325 Group coordinator for group.test is BrokerMetadata(nodeId=1008, host=u'hostname_1008', port=9092, rack=N
one)
INFO: 2016-08-17 16:50:02,783: base.py:496 Discovered coordinator 1008 for group group.test
INFO: 2016-08-17 16:50:02,785: consumer.py:268 Revoking previously assigned partitions set([]) for group group.test
INFO: 2016-08-17 16:50:02,785: base.py:288 (Re-)joining group group.test
INFO: 2016-08-17 16:50:02,787: base.py:327 Joined group 'group.test' (generation 1) with member_id kafka-python-1.3.1-6147a997-ce70-4a88-86f3-6b98215b9493
INFO: 2016-08-17 16:50:02,787: base.py:331 Elected group leader -- performing partition assignments using range
INFO: 2016-08-17 16:50:02,791: base.py:428 Successfully joined group group.test with generation 1
INFO: 2016-08-17 16:50:02,792: subscription_state.py:226 Updated partition assignment: [TopicPartition(topic=u'topic-test', partition=0), TopicPartition(topic=u'topic-test', partition=1), TopicPartition(topic=u'topic-test', partition=2), TopicPartition(topic=u'topic-test', partition=3), TopicPartition(topic=u
'topic-test', partition=4), TopicPartition(topic=u'topic-test', partition=5), TopicPartition(topic=u'topic-test', partition=6), TopicPartition(topic
=u'topic-test', partition=7)]
INFO: 2016-08-17 16:50:02,792: consumer.py:214 Setting newly assigned partitions set([TopicPartition(topic=u'topic-test', partition=6), TopicPartition(topic=u'ens
ec-url-simhash', partition=7), TopicPartition(topic=u'topic-test', partition=4), TopicPartition(topic=u'topic-test', partition=5), TopicPartition(topic=u'e
nsec-url-simhash', partition=2), TopicPartition(topic=u'topic-test', partition=3), TopicPartition(topic=u'topic-test', partition=0), TopicPartition(topic=u
'topic-test', partition=1)]) for group group.test
INFO: 2016-08-17 16:50:03,313: base.py:547 LeaveGroup request succeeded
ERROR: 2016-08-17 16:50:03,313: future.py:79 Fetch to node 1001 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,314: future.py:79 Fetch to node 1002 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,314: future.py:79 Fetch to node 1003 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,314: future.py:79 Fetch to node 1004 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,314: future.py:79 Fetch to node 1005 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,314: future.py:79 Fetch to node 1006 failed: ConnectionError: <BrokerConnection host=... port=9092>
ERROR: 2016-08-17 16:50:03,315: future.py:79 Fetch to node 1020 failed: ConnectionError: <BrokerConnection host=... port=9092>

In my case, it seems that the problem occurs when it trying to fetch message from a broker which has no more.

[secfree]

By traceback I found the reason of the error log.

It's because I called consumer.close() . The code of function close in /kafka/conn.py:

def close(self, error=None):
        """Close socket and fail all in-flight-requests.
        Arguments:
            error (Exception, optional): pending in-flight-requests
                will be failed with this exception.
                Default: kafka.common.ConnectionError.
        """
        if self._sock:
            self._sock.close()
            self._sock = None
        self.state = ConnectionStates.DISCONNECTED
        self._receiving = False
        self._next_payload_bytes = 0
        self._rbuffer.seek(0)
        self._rbuffer.truncate()
        if error is None:
            error = Errors.ConnectionError()
        while self.in_flight_requests:
            ifr = self.in_flight_requests.popleft()
            ifr.future.failure(error)

consumer.close() passes no error and so the code below is executed:

        error = Errors.ConnectionError()
        ...
        ifr.future.failure(error)

This may be misleading.

---

If i don't call consumer.close() explicitly, it may cause resource leak problems like Too many open files ... .

[secfree]

When error is None set it to Errors.ConnectionError() may also cause misleading in _process_response in /kafka/conn.py:

ifr.future.fail(error)
self.close()

[xuguoliangjj]

I have the same problem with call consumer.close(),How did you fixed this?

[secfree]

@xuguoliangjj This is not a real error, so you can just ignore it.

[xuguoliangjj]

@secfree My consumer is running normally. @dpkp Why don't you fix this？

[dpkp]

This has been fixed -- consumer.close() will log a friendly info message and send a CancelledError to in-flight futures.