abbccdda commented on a change in pull request #9170:
URL: https://github.com/apache/kafka/pull/9170#discussion_r469509680
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -193,13 +193,17 @@ private void closeAndRevive(final Map<Task,
Collection<TopicPartition>> taskWith
try {
task.suspend();
+ // we need to enforce a checkpoint that removes the corrupted
partitions
+ task.postCommit(true);
} catch (final RuntimeException swallow) {
log.error("Error suspending corrupted task {} ", task.id(),
swallow);
}
task.closeDirty();
+
+ // For active tasks pause their input partitions so we won't poll
any more records
+ // for this task until it has been re-initialized;
+ // Note, closeDirty already clears the partitiongroup for the task.
Review comment:
partition group
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -193,13 +193,17 @@ private void closeAndRevive(final Map<Task,
Collection<TopicPartition>> taskWith
try {
task.suspend();
+ // we need to enforce a checkpoint that removes the corrupted
partitions
+ task.postCommit(true);
Review comment:
I'm not sure this is 100% ensuring the snapshot gets done, as we have
branching logic in postCommit. Do you think we should just get a helper to
cleanup instead?
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
##########
@@ -58,6 +58,9 @@ static boolean checkpointNeeded(final boolean
enforceCheckpoint,
return false;
}
+ if (enforceCheckpoint)
Review comment:
Could we move this logic to the `StandbyTask` only? It is the only case
I have seen which could have null snapshot passed in, which could make this
helper assume both snapshots are not null.
```
if (oldOffsetSnapshot == null) {
return false;
}
```
----------------------------------------------------------------
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
