Handling "Fetch to node %i failed: Cancelled: ..." error

[xpyy]

Hello.

I have kafka with 3 brokers on different servers. I use kafka-python to read messages from one topic until i get timeout, like so:

def handle(self):
    consumer = KafkaConsumer(topic, **settings)
    while True:
        message = consumer.poll(timeout)
        if not message:
            break

        process_message(message)

    consumer.close()

I'm getting a lot of errors like "Fetch to node %i failed: Cancelled: ..." on 1 or 2 brokers. Debug log tells me that they actually have in flight requests with type "FetchRequest_v3".

My questions:

1. Am i do something wrong?
2. If not, how can i fix it?

[pavan538]

when ever the the socket is closed for a Node , All the pending in-flight-request will be failed with this exception. Default: kafka.errors.KafkaConnectionError.
for more insight please refer - https://github.com/dpkp/kafka-python/blob/master/kafka/conn.py BrokerConnection class clcose() - method.

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.errors.KafkaConnectionError.
        """
        if self.state is ConnectionStates.DISCONNECTED:
            if error is not None:
                log.warning('%s: Duplicate close() with error: %s', self, error)
            return
        log.info('%s: Closing connection. %s', self, error or '')
        self.state = ConnectionStates.DISCONNECTING
        self.config['state_change_callback'](self)
        self._update_reconnect_backoff()
        self._close_socket()
        self.state = ConnectionStates.DISCONNECTED
        self._sasl_auth_future = None
        self._protocol = KafkaProtocol(
            client_id=self.config['client_id'],
            api_version=self.config['api_version'])
        if error is None:
            error = Errors.Cancelled(str(self))
        while self.in_flight_requests:
            (_, future, _) = self.in_flight_requests.popleft()
            future.failure(error)
        self.config['state_change_callback'](self)

[dpkp]

Explanation above is correct -- this is expected when you close a consumer instance. Kafka FetchRequests can stay in flight for some time waiting for new messages from the broker. It is not uncommon to have in-flight requests when you close your consumer.

[joekohlsdorf]

So what is the way to handle this? Ignoring an error isn't optimal
There is no straightforward way to catch these exceptions because this is an asynchronous call and they are raised by futures.

The documentation on consumer.close() is also misleading:
Close the consumer, waiting indefinitely for any needed cleanup.
Since closing the BrokerConnection isn't waiting for in-flight requests to finish we are neither waiting nor cleaning up fully.

[jhutar]

Hello @dpkp and/or @pavan538 . Sorry for pinging in this oldish issue. What would be best way how to handle this? And also: does it mean that the doc mentioned by @joekohlsdorf above is not correct?

Close the consumer, waiting indefinitely for any needed cleanup.

[RaviTezu]

@dpkp I am facing the same issue, how can we handle these errors in a better way? I see a bunch of ERRORs in the log every time I close a consumer instance.
Thank you.

[srknzl]

@dpkp Can you please provide an explanation here, to questions above?

[Arfey]

def handle(self):
    consumer = KafkaConsumer(topic, **settings)
    message = consumer.poll(timeout)
    
    if message:
        process_message(message)

    consumer.close()

if processing of messages spend some around fetch.max.wait.ms or more time then we will not catch that errors but in another case we'll catch it each time (in case when we receive no one message).

of course i can to set time.sleep(2) before close connection but i think so it isn't correct. When I manually close connection i close my in flight requests, yes but it's not an error 😀

i propose move that log under info level (or even debug mode). #2172
@dpkp what do u think?

[geetakudumula]

Any solution on this issue? we are also facing this frequently. I know the issue is closed, but what is the solution?
Increasing the fetch.max.wait.ms is going to address this problem?

Error: "Fetch to node %i failed: Cancelled: ..."

[rigved]

I have been facing this issue as well. My speculation is that kafka/consumer/group.py:_poll_once is responsible for those extra in-flight requests:

        # If data is available already, e.g. from a previous network client
        # poll() call to commit, then just return it immediately
        records, partial = self._fetcher.fetched_records(max_records, update_offsets=update_offsets)
        if records:
            # Before returning the fetched records, we can send off the
            # next round of fetches and avoid block waiting for their
            # responses to enable pipelining while the user is handling the
            # fetched records.
            if not partial:
                futures = self._fetcher.send_fetches()
                if len(futures):
                    self._client.poll(timeout_ms=0)
            return records

So, this results in extra in-flight requests when you are closing the connection, resulting in cancellation of those requests and the displayed error.

My solution is to make an extra poll method call just before closing the connection. This solution works because I'm at the end of the current message queue (no more pending records to fetch). So, the if records if-condition gets an empty response and there are no extra in-flight requests while closing the connection.

@dpkp , could you please keep this issue open as it's still affecting users?

[usingh2buffalo]

any solution to this issue? Issue is closed but no solution was provided.

[MattMyers3]

Seeing the same issue. Any updates on a solution?