KAFKA-19128: Kafka Streams should not get offsets when close dirty

[aliehsaeedii]

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.

Reviewers: Bill Bejeck [email protected], Matthias J. Sax
[email protected]

[bbejeck]

Thanks for the PR @aliehsaeedii - this seems and like a good change overall looks good - I just have one minor comment. I'll defer final say to @mjsax since he recently made changes in this area.

[bbejeck]

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.

[mjsax]

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().

[aliehsaeedii]

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.

[mjsax]

Did I miss it, or is there no new test for this case? 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).

[mjsax]

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?

[aliehsaeedii]

I check the clean flag for accurate logs.
Let's keep the Collections.emptyMap() when close clean to keep the logic as it is?!

[aliehsaeedii]

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.

[bbejeck]

LGTM

[mjsax]

Thanks for the fix! Merged to trunk and cherry-picked to 4.0 branch.