You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2024/05/03 16:24:07 UTC
[PR] KAFKA-16665: Allow to initialize newly assigned partition's positions without allowing fetching while callback runs [kafka]
lianetm opened a new pull request, #15856:
URL: https://github.com/apache/kafka/pull/15856
Fix to allow to initialize positions for newly assigned partitions, while the onPartitionsAssigned callback is running, even though the partitions remain non-fetchable until the callback completes.
Before this PR, we were not allowing initialization or fetching while the callback was running. The fix here only allows to initialize the newly assigned partition position, and keeps the existing logic for making sure that the partition remain non-fetchable until the callback completes.
The need for this fix came out in one of the connect system tests, that attempts to retrieve a newly assigned partition position with a call to consumer.position from within the onPartitionsAssigned callback ([WorkerSinkTask](https://github.com/apache/kafka/blob/2c0b8b692071b6d436fa7e9dc90a1592476a6b17/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L715)). With this PR, we allow to make such calls (test added), which is the behaviour of the legacy consumer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16665: Allow to initialize newly assigned partition's positions without allowing fetching while callback runs [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #15856:
URL: https://github.com/apache/kafka/pull/15856
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16665: Allow to initialize newly assigned partition's positions without allowing fetching while callback runs [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15856:
URL: https://github.com/apache/kafka/pull/15856#issuecomment-2096322959
Thanks @lucasbru, conflicts solved. Good pointer about the new file for callback tests, I created https://issues.apache.org/jira/browse/KAFKA-16675 assigned to me to make sure I move the test as soon as that one gets merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16665: Allow to initialize newly assigned partition's positions without allowing fetching while callback runs [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15856:
URL: https://github.com/apache/kafka/pull/15856#issuecomment-2096266336
@lianetm some conflicts need to be resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16665: Allow to initialize newly assigned partition's positions without allowing fetching while callback runs [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15856:
URL: https://github.com/apache/kafka/pull/15856#issuecomment-2093364129
Hey @lucasbru , could you take a look?
For more context, we initially decided not to allow initializing positions while the onPartitionsAssigned callback was running, concerned about a user trying to seek in the callback, and us initializing to a different position within the poll, but that shouldn't happen because we execute the callbacks in the foreground thread, so poll will be blocked while the onPartitionsAssigned completes (so user's seek would happen, callback completes, then next poll). Actually, with the connect test we realized that we do need to allow to initialize positions, because a user may want to call .position() from within the callback, like the WorkerSinkTask does (that sys test passes locally now with this fix)
Then, this [PR](https://github.com/apache/kafka/pull/15724) was after a symptom of this same problem, but I would say we put the fix in the wrong place by not allowing reset, when in reality the issue was that we were not allowing to initialize the position in the first place. This is why with this current PR I'm kind of reverting that other PR comments, and part of the logic (not all , mainly to keep some refactoring it had).
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org