[FLINK-27204][flink-runtime] Refract FileSystemJobResultStore to execute I/O operations on the ioExecutor

[WencongLiu]

What is the purpose of the change

At present, FileSystemJobResultStore executes I/O operations through FileSystem directly. We should refract the interface of JobResultStore to make I/O operations be executed asynchronously. This would move the responsibility of I/O operation from the Dispatcher into the JobResultStore.

Brief change log

• Refract the JobResultStore interface.
• Refract all codes that calls JobResultStore.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable)

[flinkbot]

CI report:

• f59c00e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[XComp]

Hi @WencongLiu, sorry for the late response. I found time to look into you proposal now. The initial intention of FLINK-27204 was to have the async functionality being hidden in FileSystemBasedJobResultStore. The FileSystemBasedJobResultStore performs IO operations for most of the interface's methods which is considered blocking and should run in the Dispatcher's main thread (as it does right now). To achieve that, we would have to migrate all JobResultStore methods to become asynchronous:

pre-FLINK-27204 JobResultStore	post-FLINK-27204 JobResultStore*
void createDirtyResult(JobResultEntry) throws ...	CompletableFuture createDirtyResultAsync(JobResultEntry)
void markResultAsClean(JobID) throws ...	CompletableFuture markResultAsCleanAsync(JobID)
boolean hasJobResultEntry(JobID) throws ...	CompletableFuture hasJobResultEntryAsync(JobID)
boolean hasDirtyJobResultEntry(JobID) throws ...	CompletableFuture hasDirtyJobResultEntryAsync(JobID)
boolean hasCleanJobResultEntry(JobID) throws ...	CompletableFuture hasCleanJobResultEntryAsync(JobID)
Set getDirtyResults(JobID) throws ...	CompletableFuture<Set> getDirtyResultsAsync()

* The async calls in the runtime module usually have the formate CompletableFuture<?> <method-name>Async()

The FileSystemBasedJobResultStore would get an constructor parameter ioExecutor which then would be used to run the async calls. Your current proposal doesn't specify a executorService to run the CompletableFutures on. Additionally, we need to utilize the CompletableFutures whereever possible (instead of calling .get() rightaway). Calling .get() on the CompletableFuture makes the call blocking again (which is what we want to avoid).

Does that sound reasonable to you? Let me know if you have more questions.

[WencongLiu]

@XComp Thanks for your reply! 😀 I'll follow the suggestions.

[WencongLiu]

@XComp I have made a round of changes. Please take a look at it when you have time. 😃

[XComp]

Thanks, @WencongLiu . I did a pass over the change and added a few comments. PTAL

Just letting you know. I'm on vacation for the next two weeks. Therefore, don't expect any responses in that time. ...just to help you coordinating your work. :-)

[WencongLiu]

@XComp I have made a round of changes. Please take a look at it when you have time. 😄

[XComp]

Sorry for getting back to you that late, @WencongLiu . I haven't found the time till now to go through open PullRequest. Anyway, I went over the changes once more. I added a few comments. PTAL

[WencongLiu]

Thanks for your careful review 😄 @XComp. Sorry for the first round of pull request because it's a bit rough. Please take a look when you have time.

[XComp]

Thanks @WencongLiu for addressing my comments. I went over the code once more. PTAL.

I might do another pass over it to address the issue in JobMasterServiceLeadershipRunner#verifyJobSchedulingStatusAndCreateJobMasterServiceProcess next week.

[XComp]

LGTM already. I'm gonna pass over the JobMasterServiceLeadershipRunner change tomorrow.

[XComp]

Nice, we're almost there. 👍 I added a few (mostly cosmetic comments). PTAL

[WencongLiu]

Thanks for your patient review. @XComp I have added a fixup commit. PTAL.

[XComp]

Cool, thanks. We're getting closer. I have a few cosmetic comments. PTAL

[XComp]

Thanks, @WencongLiu . I only have a few cosmetic proposals. But it looks good now. :-)

One additional thing: Could you fix the TODO in JobMasterServiceLeadershipRunner:171ff in a hotfix commit. I just noticed that we forgot to clean that up 😇

[XComp]

Great. Thanks very much for your effort. One last thing is there to be fixed.

When you're done with that, you can re-organize/squash the commits properly and rebase the branch. I will do a final pass over it after this is done. 👍

[XComp]

Thanks for reorganizing the commits. In my final pass over the change, I realized that we missed one location where we should switch to the CompletableFuture handling instead of blocking the call. Sorry for missing that earlier. I marked the code snippet below. PTAL.

[XComp]

Awesome. We're done here in my opinion. Good job. 👍 I pushed some minor fixes (see comparison). The changes can be merged to master as soon as the release-1.18 branch is created.

[XComp]

CI is failing due to us not acknowledging the requirement that any state access in the Dispatcher has to happen in the main thread (to ensure sequential execution of any Dispatcher logic). Please see my comment below for further details.

[XComp]

Suggested change

	} else if (jobManagerRunnerRegistry.isRegistered(jobID)
	} else if (jobManagerRunnerRegistry.isRegistered(jobID)

Ok, that's a tricky one which I missed: The Dispatcher has one requirement: Any state access needs to happen in the main thread of the Dispatcher. There's a special implementation of JobManagerRunnerRegistry that ensures this invariant (see OnMainThreadJobmanagerRunnerRegistry). Our change (with the thenCompose being chained with the future that's returned by isInGloballyTerminalState) is going against this requirement. Why?

isInGloballyTerminalState calls jobResultStore.hasJobResultEntryAsync which executes the logic on the ioExecutor (i.e. a thread for IO operations which is not the Dispatcher's main thread) internally. The returned future is linked to this executor, i.e. any chained CompletableFuture calls will run in the same thread. The thenCompose logic is, therefore, also executed in the ioExecutor instead of the main thread.

To workaround this, we have to change the executor for the chained execution. This can be achieved by using thenComposeAsync, instead. Here we would specify the main thread executor by calling getMainThreadExecutor. One example where it's done like that is Dispatcher:619: The cleanupAsync method is executed on the ioExecutor. But the error handling has to happen in the main thread again. That's where we use handleAsync with getMainThreadExecutor.

[WencongLiu]

Thanks for your detailed explanation! 😄 I've modified the thenCompose to thenComposeAsync.

[XComp]

You didn't do a pull before adding the changes (I did a force-push to include a few minor changes previous). These changes were reverted with your most-recent push

[WencongLiu]

😼 I have added the changes in the comparison. Really sorry for this.

[XComp]

No worries. Let's see whether CI becomes 🟢 this time. 🤞

[XComp]

CI is green. 👍 Thanks again :)