Worker stuck after "Protocol out of sync"

[isamaru]

After switching to 1.4.5 from an older 1.3 version, we see some workers getting stuck with this pattern in the logs:

1. "Protocol out of sync" with no previous errors
2. "Closing connection. KafkaConnectionError: Socket EVENT_READ without in-flight-requests"
3. "[kafka.client] Node 2 connection failed -- refreshing metadata"
4. 5 minutes later, worker starts looping out infinite "Duplicate close() with error: [Error 7] RequestTimedOutError: Request timed out after 305000 ms", but with no apparent attempts to reconnect

More detailed (INFO) logs here:
kafka-python-logs.txt

We found a worker which was stuck like this for 2 days, processing no messages but not failing directly or even rebalancing the group, causing lag on its partition. The broker is running and other workers can connect to that broker in that period.
Note that Node 2 is leader for the partition which the worker is assigned to. Group coordinator is Node 1 which is why heartbeat keeps beating.

This seems to be the same thing as #1728 which wasn't completely fixed.

#1733 was about fixing one possible cause for that error (= avoid it). I believe that in our case the error is legitimate (temporary connection problems to broker).
The real issue is that the worker is not able to restore itself after this error happens and becomes stuck instead of either giving up and dying or reconnecting.
I've tried to find code responsible for reconnecting (which doesn't seem to fire) but I don't understand your codebase that well. I will continue investigating, this is important for us.

(We have deployed this on several environments and see this on both 1.0.1 and 2.1 brokers, identified by client as 1.0.0)

[isamaru]

Maybe the root cause could be an incomplete #1733 after all. I realized that the recurring timeouts will only occur if there is some in-flight-request, but there was none present before the connection was closing. Therefore I suspect a race condition between sending and storing to in-flight-request.

Ideally, addition to in-flight_requests in _send could be wrapped like so:

        with self._lock:
            correlation_id = self._protocol.send_request(request)
            log.debug('%s Request %d: %s', self, correlation_id, request)
            expect_response = request.expect_response()

            if expect_response:
                sent_time = time.time()
                assert correlation_id not in self.in_flight_requests, 'Correlation ID already in-flight!'
                self.in_flight_requests[correlation_id] = (future, sent_time)

        if not expect_response:
            future.success(None)

Afterwards, the if not conn.in_flight_requests: check in client_async.py#_poll() needs to use the same lock.

Still, missing path for reconnecting remains a problem. I haven't been able to find any code path which would reenable the connection in async_client poll loop.

[isamaru]

After further investigation, I've come to believe that it is indeed the race condition, not network problems, which are causing this. In our case we are committing manually from the main thread.
The deadlock seems to occur because "protocol out of sync" is raised and connection is closed before in-flight-requests is populated. Close() would usually terminate the futures stored in in-flight-requests but does not do that because it's not yet populated.

When close is called 5 minutes later, in-flight-requests is populated but the short-circuit in conn.py#close() does not clear the futures. Besides locking that I recommend above, we should also clean the in-flight-requests and end futures there even when short-circuiting.

[dpkp]

Interesting. Let me think a bit more about this. You went straight from 1.3.X to 1.4.5? did you ever try 1.4.4 or earlier?

[isamaru]

We've tried to upgrade to 1.4.3 last year, then reverted to 1.3.5 because we had weird cluster issues. Nothing we could pin down, specifically, and not this. No one had time to investigate properly back then.