You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2025/04/11 13:18:15 UTC
[PR] KAFKA-19128: Kafka Streams should not get offsets when close dirty [kafka]
aliehsaeedii opened a new pull request, #19450:
URL: https://github.com/apache/kafka/pull/19450
Kafka Streams calls `prepareCommit()` in `Taskmanager#closeTaskDirty()`. However, the dirty task must not get committed and therefore, prepare-commit tasks such as getting offsets should not be needed as well. The only thing needed before closing a task dirty is flushing. Therefore, separating `flush` and `prepareCommit` could be a good fix.
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #19450:
URL: https://github.com/apache/kafka/pull/19450#discussion_r2056811156
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java:
##########
@@ -179,7 +179,11 @@ public void resume() {
* or flushing state store get IO errors; such error should cause the thread to die
*/
@Override
- public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
+ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean) {
+ if (!clean) {
Review Comment:
I check the `clean` flag for accurate logs.
Let's keep the `Collections.emptyMap()` when close clean to keep the logic as it is?!
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #19450:
URL: https://github.com/apache/kafka/pull/19450#issuecomment-2798404630
Thanks for the PR -- I am wondering if it would be better to actually change `prepareCommit()` to `prepare(final boolean clean)` (instead of making `flush()` part of the task interface), and skip the call to `committableOffsetsAndMetadata()` if `clean == false` and `return null;`?
As we don't want to commit, we should not even return an empty `Map` as guard rail, and the calling code as a matter of fact ignored the return value anyway (as it should).
Btw: there is multiple places in with we "close dirty", to changing from `prepareCommit()` to `flush()` at this place won't be sufficient. -- Adding a boolean flag, also has the advantage that we won't miss any caller, as we need to update all callers.
We also don't want to lose the guard to check the task state, which we do inside `prepareCommit()` but not inside `flush()`.
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #19450:
URL: https://github.com/apache/kafka/pull/19450
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #19450:
URL: https://github.com/apache/kafka/pull/19450#issuecomment-2820937618
> Would be good to at least add one more test to `StreamsTaskTest`; something like `shouldNotGetOffsetsIfPerpareCommitDirty()` verifying that we return `null`, and ideally also that `committableOffsetsAndMetadata()` was not called (the second check is more tricky -- need some creativity I guess).
I think the 2nd check is not necessary since if `committableOffsetsAndMetadata()` is called, then `prepareCommit(bool)` would return a map and not `NULL`.
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #19450:
URL: https://github.com/apache/kafka/pull/19450#issuecomment-2829346361
Thanks for the fix! Merged to `trunk` and cherry-picked to `4.0` branch.
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #19450:
URL: https://github.com/apache/kafka/pull/19450#issuecomment-2801723666
Thanks, @bbejeck and @mjsax.
I think regarding keeping the guard to check the task state, what Matthias suggested makes more sense. So I implemented his approach.
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19450:
URL: https://github.com/apache/kafka/pull/19450#discussion_r2048191835
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java:
##########
@@ -179,7 +179,11 @@ public void resume() {
* or flushing state store get IO errors; such error should cause the thread to die
*/
@Override
- public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
+ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean) {
+ if (!clean) {
+ log.warn("Skipped preparing {} standby task with id {} for commit since the task is getting closed dirty.", state(), id);
Review Comment:
Why do we need to log this at WARN level?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -444,6 +443,10 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
//
// TODO: this should be removed after we decouple caching with emitting
flush();
+ if (!clean) {
+ log.warn("Skipped preparing {} task with id {} for commit since the task is getting closed dirty.", state(), id);
Review Comment:
as above -- should be be DEBUG ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java:
##########
@@ -179,7 +179,11 @@ public void resume() {
* or flushing state store get IO errors; such error should cause the thread to die
*/
@Override
- public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
+ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean) {
+ if (!clean) {
Review Comment:
Should this to to the end of the method? So we preserve all logging?
Or maybe even don't check `clean` flag at all, because StandbyTasks should also never return offsets, and we could just change `return Collections.emptyMap();` to `return null;` directly?
--
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-19128: Kafka Streams should not get offsets when close dirty [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #19450:
URL: https://github.com/apache/kafka/pull/19450#discussion_r2040351769
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1479,7 +1479,7 @@ private void closeTaskDirty(final Task task, final boolean removeFromTasksRegist
try {
// we call this function only to flush the case if necessary
// before suspending and closing the topology
- task.prepareCommit();
+ task.flush();
Review Comment:
The `SteamTask.prepareCommit()` set a boolean flag `hasPending TxCommit` if using EOS and the `commitNeeded` flag is `true` - is it OK to bypass that? I'm thinking so but I'd like to confirm.
--
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