MINOR: Refactor code for restoring tasks

[mjsax]

Only StreamTasks can be restored. Thus, moving restoring logic into AssignedStreamTasks class

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[mjsax]

Call for review @guozhangwang @bbejeck @vvcephei

[guozhangwang]

I like this cleanup! Thanks for doing this.

This is orthogonal to the PR, but while reviewing this I found that today it seems we always try to restore a task even if the store is not loggingEabled, what will happen if loggingEnabled is false?

[guozhangwang]

nit: @Override.

[guozhangwang]

nit: Override.

[guozhangwang]

Ditto here.

[guozhangwang]

Is this for testing only? If yes let's make it at the bottom of the class with comments.

[guozhangwang]

How about add the interface to RestoringTasks? And then in initializeNewTasks we can:

                if (!entry.getValue().initializeStateStores()) {
                    log.debug("Transitioning {} {} to restoring", taskTypeName, entry.getKey());
                    ((RestoringTasks) this).addToRestoring(entry.getValue());
                } else {
                    transitionToRunning(entry.getValue());
                }

We can also move updateRestored to that interface as well.

[mjsax]

I am not 100% sure about this.

The original motivation to add the method addToRestoring to AssignedTasks was to avoid the cast in initializeNewTasks -- it's fair request to get rid of this exception code and trade in a cast. Neither are nice solutions.

However, adding both methods to RestoringTasks does not improve the situation, as we still need the cast (we could also directly cast to AssignedStreamTasks instead) while we loose package private visibility and both methods are public if we add them to RestoringTasks.

Also RestoringTasks was added for a different purpose and adding more method would "drill a hole" into the API that we might not want. (The idea of RestoringTasks was to protect access to AssignedStreamsTask from StoreChangelogReader.)

Thoughts?

[bbejeck]

I tend to agree with @mjsax regarding the change in that adding the methods to RestoringTasks we are still left with a cast.

But IMHO the trade-off is worth it when considering the interaction with the StoreChangelogReader as the StoreChangelogReader should not have exposure to AssignedStreamsTask methods, but only the methods related to restoring as presented in the RestoringTasks interface.

Either way, it's a trade-off, but again IMHO presenting the StoreChangelogReader with an interface limited to only restoring concerns makes the most sense to me.

[mjsax]

This is orthogonal to the PR, but while reviewing this I found that today it seems we always try to restore a task even if the store is not loggingEabled, what will happen if loggingEnabled is false?

Good question. I was looking into the code a little bit. It seems, this would not really be a problem. The task is added to restoring and also for all task partitions it's added to retoringByPartition (if all stores have loggingDisabled, we would still add for input topic partitions). In AssingedStreamTasks#updateRestored we would remove it from restored via

            if (restoredPartitions.containsAll(task.changelogPartitions())) {
                transitionToRunning(task);
                it.remove();

(if all stores have loggingDisabled task.changelogPartitions() would be empty and contiansAll() would return true).

We only don't clean up restoringByPartition it seems... Thoughts?

[mjsax]

Call for review @bbejeck @vvcephei

Please consider and comment: #5768 (comment)

[bbejeck]

Overall I don't have any further comments beyond what @guozhangwang has presented already. The cleanup looks good to me.

I've left my comments for #5768 (comment) in-line.

[bbejeck]

Can we run the EOS system tests as one last check before merging?

[guozhangwang]

retest this please

[guozhangwang]

Can we run the EOS system tests as one last check before merging?

+1

[guozhangwang]

We only don't clean up restoringByPartition it seems... Thoughts?

Thanks for looking into this! I think it's okay to leave it in restoringByPartition since we do not actively remove entries from it until cleanup anyways, and we only fetch entries from it is restoringTaskFor which is not affected. So keeping this map entry does not hurt much.

[guozhangwang]

LGTM. @mjsax please feel free to merge if the EOS system test passed as well.

[mjsax]

Remove the methods from RestoringTasks interface, as discussed, and rebased. If not further comments come in, I'll merge after build passed.

[guozhangwang]

Checkstyle failure:

21:32:10 FAILURE: Build failed with an exception.
21:32:10 
21:32:10 * What went wrong:
21:32:10 Execution failed for task ':streams:checkstyleMain'.

All tests passed.

[guozhangwang]

Remove the methods from RestoringTasks interface, as discussed, and rebased. If not further comments come in, I'll merge after build passed.

Did you run EOS system test to verify it passed?

[mjsax]

Fixed checkstyle error.

Triggered system test: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2085/

[mjsax]

Some system tests failed because of unrelated bug. Filed #5928 to fix the system tests. Can rebase this PR after fix is merged.

Build timed out. Will retrigger Jenkins after rebase.

[mjsax]

Rebased to pick up system test fixes from #5928. Triggered system tests again: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2090/

[mjsax]

Rebased to pick up system test fix. Re-triggered system tests: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2097/

[mjsax]

System test passed. Merging.