AssertionError: RecordAccumulator is closed

[LplusKira]

Hi guys, we got some errors like

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.6/threading.py", line 1182, in run
    self.function(*self.args, **self.kwargs)
  File "/home/po-kai.chang/.local/lib/python3.6/site-packages/kafka_logger/handlers.py", line 295, in flush
    self.producer.send(self.kafka_topic_name, log)
  File "/home/po-kai.chang/.local/lib/python3.6/site-packages/kafka/producer/kafka.py", line 603, in send
    estimated_size=message_size)
  File "/home/po-kai.chang/.local/lib/python3.6/site-packages/kafka/producer/record_accumulator.py", line 247, in append
    assert not self._closed, 'RecordAccumulator is closed'
AssertionError: RecordAccumulator is closed

Any ideas?

After commenting out

kafka-logging-handler/kafka_logger/handlers.py

Line 349 in 633980c

self.producer.close()

There's no such error (and no other errors seen). Maybe this is related?

our handler's config is like

    bootstrap_server = "someServer"
    topic = "someTopic"
    kafka_producer_args = {
        'sasl_mechanism': "SCRAM-SHA-512"
        'sasl_plain_username': "someUserName",
        'sasl_plain_password': "somePwd",
        'acks': 0
    }
    kafka_log_handler = KafkaLoggingHandler(bootstrap_server,
                                            topic,
                                            security_protocol="SASL_SSL",
                                            ssl_cafile="someFile",
                                            kafka_producer_args=kafka_producer_args)

[bobvt]

we are getting the same exception and stack trace. it its intermittent and happens only when running in batch. its an aws sagemaker batch process. we are also using gthread; not sure if that would cause a problem with the daemon thread spawned in the handler. any thoughts???