Deadlock between consumer and heartbeat thread during coordinator lookup

[zhgjun]

consumer dead after broker restart, it cannot receive message and maybe dead . and I got the stack trace:

========================================================================
====                         Green Threads                          ====
========================================================================
------                        Green Thread                        ------

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/eventlet/green/thread.py:41 in __thread_body
    `func(*args, **kwargs)`

/usr/lib64/python2.7/threading.py:784 in __bootstrap
    `self.__bootstrap_inner()`

/usr/lib64/python2.7/threading.py:811 in __bootstrap_inner
    `self.run()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/coordinator/base.py:964 in run
    `self._run_once()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/coordinator/base.py:999 in _run_once
    `self.coordinator._client.poll(timeout_ms=0)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/client_async.py:560 in poll
    `responses.extend(self._fire_pending_completed_requests())`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/client_async.py:659 in _fire_pending_completed_requests
    `future.success(response)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/future.py:36 in success
    `self._call_backs('callback', self._callbacks, self.value)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/future.py:79 in _call_backs
    `f(value)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/coordinator/base.py:716 in _handle_group_coordinator_response
    `with self._client._lock, self._lock:`

/usr/lib64/python2.7/threading.py:173 in acquire
    `rc = self.__block.acquire(blocking)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/eventlet/semaphore.py:113 in acquire
    `hubs.get_hub().switch()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/eventlet/hubs/hub.py:294 in switch
    `return self.greenlet.switch()`



------                        Green Thread                        ------
    `messages = self.consumer.poll(timeout * 1000.0)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/consumer/group.py:614 in poll
    `records = self._poll_once(remaining, max_records)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/consumer/group.py:634 in _poll_once
    `self._coordinator.poll()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/coordinator/consumer.py:262 in poll
    `self.ensure_coordinator_ready()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/coordinator/base.py:259 in ensure_coordinator_ready
    `self._client.poll(future=future)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/client_async.py:556 in poll
    `self._poll(timeout)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/client_async.py:574 in _poll
    `ready = self._selector.select(timeout)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/kafka/vendor/selectors34.py:342 in select
    `r, w, _ = self._select(self._readers, self._writers, [], timeout)`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/eventlet/green/select.py:86 in select
    `return hub.switch()`

/opt/cloud/services/network-agent/venv/lib/python2.7/site-packages/eventlet/hubs/hub.py:294 in switch
    `return self.greenlet.switch()`

anyone help? @dpkp @tvoinarovskyi @jeffwidman , any ideas?

[tvoinarovskyi]

Do you spawn several green threads for consumption?

[tvoinarovskyi]

Please post some example similar to your processing code

[licy121]

I think the reason is that the poll method of the heartbeat thread is not locked,
self.coordinator._client.poll(timeout_ms=0)
causing the response of the main thread group_coordinator_request to be received by the heartbeat thread, so the main thread is stuck in an infinite loop to wait for furture.is_done.

    future = self.lookup_coordinator()
    self._client.poll(future=future) 

    def poll(self, timeout_ms=None, future=None):
        """Try to read and write to sockets.

        This method will also attempt to complete node connections, refresh
        stale metadata, and run previously-scheduled tasks.

        Arguments:
            timeout_ms (int, optional): maximum amount of time to wait (in ms)
                for at least one response. Must be non-negative. The actual
                timeout will be the minimum of timeout, request timeout and
                metadata timeout. Default: request_timeout_ms
            future (Future, optional): if provided, blocks until future.is_done

        Returns:
            list: responses received (can be empty)
        """
        if future is not None:
            timeout_ms = 100
        elif timeout_ms is None:
            timeout_ms = self.config['request_timeout_ms']
        elif not isinstance(timeout_ms, (int, float)):
            raise RuntimeError('Invalid type for timeout: %s' % type(timeout_ms))

        # Loop for futures, break after first loop if None
        responses = []
        while True:
            with self._lock:

                # Attempt to complete pending connections
                for node_id in list(self._connecting):
                    self._maybe_connect(node_id)

                # Send a metadata request if needed
                metadata_timeout_ms = self._maybe_refresh_metadata()

                # If we got a future that is already done, don't block in _poll
                if future is not None and future.is_done:
                    timeout = 0
                else:
                    idle_connection_timeout_ms = self._idle_expiry_manager.next_check_ms()
                    timeout = min(
                        timeout_ms,
                        metadata_timeout_ms,
                        idle_connection_timeout_ms,
                        self.config['request_timeout_ms'])
                    timeout = max(0, timeout / 1000)  # avoid negative timeouts

                self._poll(timeout)

            # called without the lock to avoid deadlock potential
            # if handlers need to acquire locks
            responses.extend(self._fire_pending_completed_requests())

            # If all we had was a timeout (future is None) - only do one poll
            # If we do have a future, we keep looping until it is done
            if future is None or future.is_done:
                break

        return responses

So I think it can be solved by locking the poll method of the heartbeat thread like this:

with self.coordinator._client._lock, self.coordinator._lock:
     self.coordinator._client.poll(timeout_ms=0)

@dpkp @tvoinarovskyi Help me confirm?

[zhgjun]

Please post some example similar to your processing code

Please post some example similar to your processing code
I use the kafka-python 1.4.4(or master branch).when i start about 600 processes which code like below(they are subscribe same topic, but not same group).when restart kafka broker,I get some of them dead,not receive message any more.

...
consumer = kafka.KafkaConsumer(topics, group_id=_group_id,...)
while True:
    msg = consumer.poll(timeout_ms=5)
    print msg
    time.sleep(1)

I read the code, the situation is probable.

1.in https://github.com/dpkp/kafka-python/blob/master/kafka/coordinator/base.py#L258 , lookup_coordinator() will send _send_group_coordinator_request(), and will do self._client.poll(future=future) to handle this response. but in self._client.poll() have no lock, heartbeat thread can do poll too .

2._fire_pending_completed_requests https://github.com/dpkp/kafka-python/blob/master/kafka/client_async.py#L576 which out the self._lock. if heartbeat thread got the coordinator_request's response which main thread send and when handle the response in https://github.com/dpkp/kafka-python/blob/master/kafka/coordinator/base.py#L666 , it need the lock that main thread had got in https://github.com/dpkp/kafka-python/blob/master/kafka/coordinator/base.py#L246 .

3. finally the main thread never end up the _client.poll() loop.The only condition of end the loop is future.is_done because the future cannot be is_done forever (heartbeat thread is acquiring lock to handle this). And this turn out as what I had meet.

so I think this is the reason.

[dpkp]

Agree w/ zhgjun -- explanation makes sense and seems like a potential source of deadlock. Needs a fix. I'd like to think a little bit about the best path forward. The current heartbeat thread design is not my favorite.

[licy121]

Agree w/ zhgjun -- explanation makes sense and seems like a potential source of deadlock. Needs a fix. I'd like to think a little bit about the best path forward. The current heartbeat thread design is not my favorite.

Hello, @dpkp @tvoinarovskyi , I saw your modification. May I have any questions about my modification in heartbeat thread like this?

with self.coordinator._client._lock, self.coordinator._lock:
     self.coordinator._client.poll(timeout_ms=0)

I mainly consider the following two points. The first is that the poll method of the Java client heartbeat thread requires two locks,I am worried that if it is completely released, there may be deadlocks in other places. In addition, the _fire_pending_completed_requests method of the Java client is not in the lock status.
I think the reason for the deadlock is that the heartbeat thread should not poll this time, and the fundamental solution is to let the heartbeat thread acquire self.coordinator._lock, so I made such a modification, please look at it if you have any questions，thank you!

[dpkp]

The way we are dealing with deadlock right now is to try to enforce a global lock order of client -> coordinator. As long as everything is playing by these rules, acquiring the client lock by itself should not be a problem. The problem will occur when code acquires the coordinator lock alone, and then later attempts to acquire the client lock. We should make sure this never happens.

In this case, having the heartbeat thread call self.coordinator._client.poll(timeout_ms=0) without any locks should be fine. It does not hold the coordinator lock and therefore can't cause deadlock (though it can be affected by it if some other code path tries to acquire locks out of order). Inside client.poll() the client lock will always be acquired first, which is the correct lock order.

[dpkp]

Are you still seeing this issue on master w the latest deadlock fix?