Deadlock in Python 3.7 when KafkaProducer closed as part of logging reconfiguration

[nickwilliams-eventbrite]

I've reduced this to as simple a reproduction as possible, so bear with me a bit, and if anything doesn't make sense, just ask...

Using the latest 1.4.7 version of Kafka-Python.

In earlier versions of Python, if you call logging.config.dictConfig and then later call it again to re-configure logging, the close method of the previously-configured logging handlers was not called to clean up the abandoned handlers. This was a bug in Python that was fixed in Python 3.7. Now, when dictConfig is called a second time, the previous handlers are all cleaned up by having their close method called. If a Kafka producer is used in a logging handler and its close method is called within the handler's close method, it causes a deadlock that prevents program execution from continuing.

First, here's a simple use-case that works on Python 2.7, 3.5, and 3.7:

def replicate():
    sys.stdout.write('Creating producer...\n')
    sys.stdout.flush()
    producer = KafkaProducer(
        acks=1,
        bootstrap_servers=['kafka:9092'],
        client_id='analytics_producer',
        compression_type='gzip',
        linger_ms=100,
        max_block_ms=1000,
        request_timeout_ms=3000,
    )
    sys.stdout.write('Closing producer...\n')
    sys.stdout.flush()
    producer.close(timeout=2)

>>> replicate()
Creating producer...
Closing producer...
>>> replicate()
Creating producer...
Closing producer...
>>> replicate()
Creating producer...
Closing producer...

Each call to replicate completes in < 1 second. As you can see, this working use case does not involve logging. But once you involve logging as below, things break:

import logging
import logging.config
import sys
from kafka.producer import KafkaProducer


class KafkaAnalyticsLoggingHandler(logging.Handler):
    def __init__(self):
        super(KafkaAnalyticsLoggingHandler, self).__init__()
        sys.stdout.write('Creating producer...\n')
        sys.stdout.flush()
        self._producer = KafkaProducer(
            acks=1,
            bootstrap_servers=['kafka:9092'],
            client_id='analytics_producer',
            compression_type='gzip',
            linger_ms=100,
            max_block_ms=1000,
            request_timeout_ms=3000,
        )

    def emit(self, record):
        return

    def close(self):
        sys.stdout.write('Closing producer...\n')
        sys.stdout.flush()
        self._producer.close(timeout=2)
        super(KafkaAnalyticsLoggingHandler, self).close()


def replicate():
    config = {'version': 1, 'handlers': {'analytics': {'level': 'INFO', 'class': '__main__.KafkaAnalyticsLoggingHandler'}}, 'loggers': {'analytics': {'handlers': ['analytics'], 'level': 'INFO'}}, 'disable_existing_loggers': False}
    logging.config.dictConfig(config)
    logging.config.dictConfig(config)

>>> replicate()
Creating producer...
Closing producer...
^CTraceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 4, in replicate
  File "/usr/local/lib/python3.7/logging/config.py", line 799, in dictConfig
    dictConfigClass(config).configure()
  File "/usr/local/lib/python3.7/logging/config.py", line 535, in configure
    _clearExistingHandlers()
  File "/usr/local/lib/python3.7/logging/config.py", line 272, in _clearExistingHandlers
    logging.shutdown(logging._handlerList[:])
  File "/usr/local/lib/python3.7/logging/__init__.py", line 2034, in shutdown
    h.close()
  File "<stdin>", line 20, in close
  File "/usr/local/lib/python3.7/site-packages/kafka/producer/kafka.py", line 493, in close
    self._sender.join()
  File "/usr/local/lib/python3.7/threading.py", line 1044, in join
    self._wait_for_tstate_lock()
  File "/usr/local/lib/python3.7/threading.py", line 1060, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt

In this case, Closing producer... appears and then it freezes. No prompt, no CPU usage, just total deadlock. Only Ctrl+C can get you out of it, and then you see the stack trace.

This only happens in Python 3.7, because that's when the logging config bug was fixed. In Python 2.7 and 3.5, there are no deadlocks and you see the output below. Note that it never closes the producer, because close is never called on the handler:

>>> replicate()
Creating producer...
Creating producer...
>>> replicate()
Creating producer...
Creating producer...
>>> replicate()
Creating producer...
Creating producer...

I think the first, working use case demonstrates that this isn't a timing issue. It's possible to rapidly close a producer after instantiating it without errors or deadlocks. The problem only occurs when the Producer's close method is called from within a logging handler's close method during logging reconfiguration. Critically, it does not happen when logging is shut down at the end of the application! Demonstration (you need a fresh Python shell to attempt this):

>>> config = {'version': 1, 'handlers': {'analytics': {'level': 'INFO', 'class': '__main__.KafkaAnalyticsLoggingHandler'}}, 'loggers': {'analytics': {'handlers': ['analytics'], 'level': 'INFO'}}, 'disable_existing_loggers': False}
>>> logging.config.dictConfig(config)
Creating producer...
>>> exit()
Closing producer...
nwilliams $

I'm certain this has something to do with Python logging's locks interfering with the thread locks of the Kafka sender somehow, but I don't begin to understand how. 🤔

[nickwilliams-eventbrite]

Another note: I cannot replicate this problem using a simple thread without involving the Kafka-Python library, so the bug does appear to for sure be somewhere within this library and not in Python itself:

import logging
import logging.config
import sys
import threading
import time


class DoNothingThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self._running = False
        self.daemon = True

    def run(self):
        self._running = True
        while self._running:
            time.sleep(0.5)

    def close(self):
        self._running = False
        self.join()


class KafkaAnalyticsLoggingHandler(logging.Handler):
    def __init__(self):
        super(KafkaAnalyticsLoggingHandler, self).__init__()
        sys.stdout.write('Creating producer...\n')
        sys.stdout.flush()
        self._producer = DoNothingThread()
        self._producer.start()

    def emit(self, record):
        return

    def close(self):
        sys.stdout.write('Closing producer...\n')
        sys.stdout.flush()
        self._producer.close()
        super(KafkaAnalyticsLoggingHandler, self).close()


def replicate():
    config = {'version': 1, 'handlers': {'analytics': {'level': 'INFO', 'class': '__main__.KafkaAnalyticsLoggingHandler'}}, 'loggers': {'analytics': {'handlers': ['analytics'], 'level': 'INFO'}}, 'disable_existing_loggers': False}
    logging.config.dictConfig(config)
    logging.config.dictConfig(config)

>>> replicate()
Creating producer...
Closing producer...
Creating producer...
>>> replicate()
Closing producer...
Creating producer...
Closing producer...
Creating producer...
>>> exit()
Closing producer...
nwilliams $

[jeffwidman]

Thanks for the fantastic bug report. I don't have time just yet to look at this, but wanted to let you know I saw it and will try to take a look at some point here.

[dpkp]

Every logging call acquires a single logging lock, and I believe the same lock is held during logging config and close. The reason there is deadlock here is because Sender.run makes calls to logging (via log.debug()) and is blocked waiting for the lock while the logging handler is blocked waiting for the sender to complete. One way to replicate this in your minimal code is to add logging.info('deadlock!') as the first line of DoNothingThread.run() .

There's not a lot we can do explicitly to avoid this -- but it does seem like a simple bug that producer.close(timeout=2) can still block forever. I put up a PR to change that and honor the timeout.

[nickwilliams-eventbrite]

Awesome! I see you merged that about five days ago. Any thoughts on what version that fix might be released in?

[jeffwidman]

@nickwilliams-eventbrite https://github.com/dpkp/kafka-python/releases/tag/2.0.0 🎉

I'm going to close as this appears to be mostly fixed, and there's not much we can do about the remaining bit, per Dana's comment above.