nicktelford commented on code in PR #15264: URL: https://github.com/apache/kafka/pull/15264#discussion_r1473153338
########## streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java: ########## @@ -1097,10 +1102,10 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) { for (final Task task : commitNeededActiveTasks) { if (!dirtyTasks.contains(task)) { try { - // for non-revoking active tasks, we should not enforce checkpoint - // since if it is EOS enabled, no checkpoint should be written while - // the task is in RUNNING tate - task.postCommit(false); + // we only enforce a checkpoint if the transaction buffers are full + // to avoid unnecessary flushing of stores under EOS + final boolean enforceCheckpoint = maxUncommittedStateBytes > -1 && tasks.approximateUncommittedStateBytes() >= maxUncommittedStateBytes; + task.postCommit(enforceCheckpoint); Review Comment: Same as above. That said, I'm not convinced this particular change is necessary. This code is run during rebalance, and AFAIK no records will be processing during a rebalance; it shouldn't be possible for the transaction buffer to have increased beyond the last time we checked in the `StreamThread` main run-loop (aka. `StreamThread#maybeCommit`). Worst case scenario, if we leave this as `task.postCommit(false)`, and the state stores aren't flushed during a rebalance, if the transaction buffer is already close to exceeding its capacity, it will be detected during the next iteration of the run-loop. -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org