ValueError: I/O operation on closed epoll object with kafka-python >= 1.4.2

[robooo]

I've started to get this exception after updating library. No code changes, I'm doing only
Connect consumer
Subscribe
Read messages
Close

I'm always reading latest messages with the same group_id

Do you have any idea what could go wrong?
Note: With 1.4.1 version I get no exceptions

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.6/site-packages/kafka/coordinator/base.py", line 914, in run
    self._run_once()
  File "/usr/local/lib/python3.6/site-packages/kafka/coordinator/base.py", line 952, in _run_once
    future = self.coordinator.lookup_coordinator()
  File "/usr/local/lib/python3.6/site-packages/kafka/coordinator/base.py", line 284, in lookup_coordinator
    future = self._send_group_coordinator_request()
  File "/usr/local/lib/python3.6/site-packages/kafka/coordinator/base.py", line 648, in _send_group_coordinator_request
    elif not self._client.ready(node_id, metadata_priority=False):
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 382, in ready
    self._maybe_connect(node_id)
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 368, in _maybe_connect
    conn.connect()
  File "/usr/local/lib/python3.6/site-packages/kafka/conn.py", line 344, in connect
    self.config['state_change_callback'](self)
  File "/usr/local/lib/python3.6/site-packages/kafka/util.py", line 159, in __call__
    return self.method()(self.target(), *args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 289, in _conn_state_change
    self._selector.register(conn._sock, selectors.EVENT_WRITE)
  File "/usr/local/lib/python3.6/selectors.py", line 412, in register
    self._epoll.register(key.fd, epoll_events)
ValueError: I/O operation on closed epoll object

[jeffwidman]

What is your code?

What flavor of linux are you using?

[robooo]

I've already switched to confluent kafka but my code was like

def connect_consumer(self,
                         server="127.0.0.1",
                         port="9092",
                         group_id="test_group",
                         enable_auto_commit=False,
                         auto_offset_reset="earliest",
                        ):
        logger = logging.getLogger("kafka")
        logger.addHandler(logging.StreamHandler(sys.stdout))
        logger.setLevel(logging.DEBUG)


        self.__consumer = KafkaConsumer(bootstrap_servers='{}:{}'.format(server, port),
                                        group_id=group_id,
                                        auto_offset_reset=auto_offset_reset,
                                        enable_auto_commit=enable_auto_commit
                                       )
def subscribe(self, topic):
        return self.__consumer.subscribe(topic)

def set_offset_to_end(self):
        self.__consumer.seek_to_end()

def get_messages_gen(self, max_records, timeout, data_type=None):
        self.__consumer.poll()
        self.set_offset_to_end()

        messages = self.__consumer.poll(timeout_ms=timeout, max_records=max_records).values()
        for attr in messages:
            for value in attr:
                data = value.value
                if data_type == 'binary':
                    yield data
                else:
                    yield self.decoded_utf(data)

[dpkp]

This happened on close, correct? I believe the issue is a race condition between closing the consumer and closing the heartbeat thread. We currently do not block waiting for the heartbeat thread to fully shutdown, so it is possible for it to attempt to make client calls / broker connections after the consumer and client are shutdown. That is what the stacktrace appears to show.

    self._epoll.register(key.fd, epoll_events)
ValueError: I/O operation on closed epoll object

suggests that the client selector has already been closed. This only happens in KafkaClient.close(), which would only happen here in KafkaConsumer.close(). So I think the solution is to block until the heartbeat thread terminates before continuing with the consumer / client shutdown. I can put up a quick PR for that.

[jeffwidman]

Yeah, this seems valid issue... I recall also seeing an exception on consumer.close(), likely this issue although at the time I was focused on another problem and forgot to grab the stack trace.