Weird error after bad producer initialization

[pecalleja]

Hi,
I'm getting this weird error after the GC tries to clean the memory and trigger the __del__ method of KafkaClient. When the producer is not successfully initialized, the property _closed is not ready in the KafkaClient object

this snippet if to illustrate the point:

from kafka import KafkaProducer
import os
import gc

def test_bad_kafka_initialization():
    connect_str = os.getenv('BROKER_URL')
    try:
        producer = KafkaProducer(bootstrap_servers=connect_str)
    except Exception as e:
        gc.collect()
        assert isinstance(e, TypeError)
        raise e

if the BROKER_URL env variable is not defined is pretty normal raise a TypeError exception but after that, another ignored exception shows up :

Exception ignored in: <function KafkaClient.__del__ at 0x10ecc04c0>
Traceback (most recent call last):
  File "..../python3.8/site-packages/kafka/client_async.py", line 443, in __del__
  File "..../python3.8/site-packages/kafka/client_async.py", line 417, in _close
AttributeError: 'KafkaClient' object has no attribute '_closed'

this happens because in KafkaClient.__init__ are properties defined after too after all initialization:

 self.cluster = ClusterMetadata(**self.config)   -> here the initialization broke
 .....
 self._closed = False
 self._wake_r, self._wake_w = socket.socketpair()
 self._selector = self.config['selector']()