Allow the coordinator to auto-commit for api_version==0.8.1

[justecorruptio]

Raised once in #1594

We're using 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.

If we allow for 0.8.1 in ConsumerCoordinator.poll(), then everything seems fine. Messages are yielded, and commits are written to ZK.

---

This change is 

[jeffwidman]

This seems safe to me, although it's weird that zookeeper commits work at all for KafkaConsumer!

0.8.2 is required for supporting commits to kafka's __consumer_offsets topic, so that's probably why this requirement is in there.

You didn't see any other unexpected issues when lowering this requirement? I was half-expecting something else to blow up if it was also trying to commit to kafka...

[jeffwidman]

Also, I don't remember this far back... in 0.8.1, was the client issuing the commits to zookeeper, or was it still issuing the commits to the broker and then the broker piped them through to zookeeper? If the latter, that would explain why this works... and in fact we may want to just remove the requirement altogether (at least at this point in the code)

[justecorruptio]

Mmm, I'm not familiar with anything past 0.8.1 actually. But I haven't seen any issues, and 1.3.5 was working with this setup.

Yes, 0.8.1 is where the writes to zookeeper are from the broker instead of the client.

[jeffwidman]

Yes, 0.8.1 is where the writes to zookeeper are from the broker instead of the client.

If that's the case, I'd prefer to simply change this to if self.group_id is None: and drop the version check altogether.

If you do that, I'll happily merge... this is very low risk as it won't change behavior for anyone > 0.8.1, and that's the vast majority of our users.

[justecorruptio]

aye aye!

[jeffwidman]

Thank you again!