auto commit never executed with api_version 0.8.1

[gaetano-guerriero]

Hello,
I'm using KafkaConsumer with api_version=(0, 8, 1) and enable_auto_commit=True.
I need to use api_version 0.8.1 since the consumer I'm migrating from uses OffsetCommitRequest version 0 and so stores offsets on zookeeper.

The offsets are not periodically committed as expected. From what I can understand this is due to early exit in coordinator.ConsumerCoordinator.poll()

[jeffwidman]

Currently, KafkaConsumer doesn't support writing offsets to Zookeeper: https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html#kafka.KafkaConsumer.commit

You'll need to use the old / deprecated SimpleConsumer to commit to zookeeper.

[justecorruptio]

Hi hi,

We're doing this exact same thing ie. api_version=(0, 8, 1) and enable_auto_commit=True. This had worked in kafka-python==1.3.5 but fails now in kafka-python==1.4.6.

In 1.3.5 even though the documentation suggested that commits to ZooKeeper is not supported, we could see the commit reflected upon inspection with zkCli.

In 1.4.6 though, KafkaConsumer just doesn't function with the above configuration; because the early exit in ConsumerCoordinator.poll() prevents the coordinator from updating next_auto_commit_deadline. This stale value, in turn, causes KafkaConsumer._message_generator() to preempt the iteration of self._fetcher until consumer_timeout is reached.

Interestingly, if we simply allow for 0.8.1 in ConsumerCoordinator.poll() ie. if self.group_id is None or self.config['api_version'] < (0, 8, 1): then everything seems fine. Messages are yielded, and commits are written to ZK.

I must be missing something... Can anyone shed light to this?
Thanks thanks!