KAFKA-10166: always write checkpoint before closing an (initialized) task

[ableegoldman]

This should address at least some of the excessive TaskCorruptedExceptions we've been seeing lately. Basically, at the moment we only commit tasks if commitNeeded is true -- this seems obvious by definition. But the problem is we do some essential cleanup in postCommit that should always be done before a task is closed:

1. clear the PartitionGroup
2. write the checkpoint

2 is actually fine to skip when commitNeeded = false with ALOS, as we will have already written a checkpoint during the last commit. But for EOS, we only write the checkpoint before a close -- so even if there is no new pending data since the last commit, we have to write the current offsets. If we don't, the task will be assumed dirty and we will run into our friend the TaskCorruptedException during (re)initialization.

To fix this, we should just always call prepareCommit and postCommit at the TaskManager level. Within the task, it can decide whether or not to actually do something in those methods based on commitNeeded.

One subtle issue is that we still need to avoid checkpointing a task that was still in CREATED, to avoid potentially overwriting an existing checkpoint with uninitialized empty offsets. Unfortunately we always suspend a task before closing and committing, so we lose the information about whether the task as in CREATED or RUNNING/RESTORING by the time we get to the checkpoint. For this we introduce a special flag to keep track of whether a suspended task should actually be checkpointed or not

[ableegoldman]

This was another "sort-of bug": if we hit an exception in handleRevocation we wouldn't finish committing the active tasks, so commitNeeded could still be true. But of course, if we hit an exception earlier, we would have thrown it up to ConsumerCoordinator which would only save the first exception, so this didn't really do anything

[ableegoldman]

We can actually simplify the standby task shutdown a LOT

[guozhangwang]

Makes sense.

[guozhangwang]

I made a pass over the code, overall it LGTM.

While working on another PR I realized that the stateMgr.flush actually does not need to be in prepareCommit, and postCommit is sufficient; and in that case we can just optionally call postCommit upon each commit. Anyways, just a quick FYI for something that's out of this scope.

[guozhangwang]

If we are not committing these tasks, should we call their postCommit?

[guozhangwang]

Makes sense.

[guozhangwang]

@ableegoldman please lmk when you want to trigger jenkins builds on this PR.

[abbccdda]

retest this

[abbccdda]

test this

[vvcephei]

Test this please

[vvcephei]

Ok to test

[vvcephei]

Retest this please

[vvcephei]

Oh, yeah... the magic touch ;)

[ableegoldman]

Seems like all the Topology testDriver tests failed, but I got a green build running locally. Do they not run with ./gradlew streams:test?

[guozhangwang]

Seems like all the Topology testDriver tests failed, but I got a green build running locally. Do they not run with ./gradlew streams:test?

They are included in streams:test. Maybe try to rebase the branch and see if there's any missing committs?

[vvcephei]

Test this please

[vvcephei]

Test this please

[ableegoldman]

Java 8 and 14 builds passed, Java 11 build failed with...zero failures?

[abbccdda]

12:23:50 
12:23:50 * What went wrong:
12:23:50 Execution failed for task ':streams:unitTest'.
12:23:50 > Process 'Gradle Test Executor 9' finished with non-zero exit value 134
12:23:50   This problem might be caused by incorrect test process configuration.
12:23:50   Please refer to the test execution section in the User Manual at https://docs.gradle.org/6.5/userguide/java_testing.html#sec:test_execution
12:23:50 ```

[guozhangwang]

Cherry-picked to 2.6 since it is a blocker, cc @rhauch