Bug fix: KafkaConsumer.position() #577
Conversation
Method KafkaConsumer.position() was not refreshing position correctly when calling self._update_fetch_positions().
| @@ -462,7 +462,7 @@ def _poll_once(self, timeout_ms): | |||
| return records | |||
|
|
|||
| self._fetcher.init_fetches() | |||
| self._client.poll(timeout_ms) | |||
| self._client.poll(timeout_ms, sleep=True) | |||
There was a problem hiding this comment.
This is a substantial change - why? The current consumer design expects poll() to return immediately if there is nothing to block on.
There was a problem hiding this comment.
When a consumer is running that doesn't currently have anything to do it will print out a constant stream of the following error when poll is called in a loop (which I assume is the normal method of consuming endlessly):
"_poll called with a non-zero timeout and sleep=False -- but there was nothing to do. This can cause high CPU usage during idle."
This is similar to issue #494 that had been fixed but it seems that commit b8c2097 broke it again. I feel that it is unneeded work for every client to check on the partition assignments before every call to poll to make sure that it is non-empty. For the client, there is little difference between receiving no data because there are no partitions currently assigned and receiving no data because the topic has no new messages. So allowing poll to sleep if a timeout is specified helps simplying the usage of the library.
There was a problem hiding this comment.
Thanks for the thoughts. Can you remove from this PR? This is a separate issue (and also I believe the incorrect fix for the bug you described).
|
You are right, it is a different issue that accidentally got added to the pull request because it was tracking my master. I have removed it. The poll() sleep issue is in issue #572. |
Method KafkaConsumer.position() was not refreshing position correctly
when calling self._update_fetch_positions().