ableegoldman commented on a change in pull request #8924:
URL: https://github.com/apache/kafka/pull/8924#discussion_r445212765
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -515,17 +517,20 @@ private void writeCheckpoint() {
stateMgr.checkpoint(checkpointableOffsets());
}
- /**
- * You must commit a task and checkpoint the state manager before closing
as this will release the state dir lock
- */
- private void close(final boolean clean) {
- if (clean && commitNeeded) {
- // It may be that we failed to commit a task during
handleRevocation, but "forgot" this and tried to
- // closeClean in handleAssignment. We should throw if we detect
this to force the TaskManager to closeDirty
+ private void validateClean() {
+ // It may be that we failed to commit a task during handleRevocation,
but "forgot" this and tried to
+ // closeClean in handleAssignment. We should throw if we detect this
to force the TaskManager to closeDirty
+ if (commitNeeded) {
log.debug("Tried to close clean but there was pending uncommitted
data, this means we failed to"
+ " commit and should close as dirty instead");
throw new TaskMigratedException("Tried to close dirty task as
clean");
}
+ }
+
+ /**
+ * You must commit a task and checkpoint the state manager before closing
as this will release the state dir lock
+ */
+ private void close(final boolean clean) {
Review comment:
This diff turned out a bit awkward, basically I just factored this check
out into a separate method that we should call at the beginning of both flavors
of clean close
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
