[FLINK-11813] Introduces basic JobManagerRunner implementation #3

XComp · 2021-12-09T17:15:41Z

What is the purpose of the change

Two goals are covered by this PR:

Consolidation of the cleanup phase (moving cleanup logic into GlobalCleanupStage) and providing a localCleanup method in the Dispatcher with now retry mechanism involved, yet!
Implements a basic JobManagerRunner class that only instantiates the checkpoint-related components and implements close logic for them. No retry mechanism is applied, yet.

Follow-ups:

RetryStategy needs to be integrated
The two classes implementing JobManagerRunner (i.e. JobMasterServiceLeadershipRunner and CheckpointJobDataCleanupRunner) have to be consolidated. General logic (the leader election) should be moved into an abstract class
We might want to integrate the ExecutionGraphInfoStore in the CheckpointJobDataCleanupRunner as a fallback.

Brief change log

See individual commits and their messages for descriptions.

Verifying this change

[TODO] This PR contains only a prototype. Additional efforts (like implementing tests, adding documentation) need to be applied.

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

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

Documentation

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

dmvk · 2021-12-14T12:54:15Z

flink-core/src/main/java/org/apache/flink/core/fs/local/LocalFileSystem.java

    private boolean delete(final File f) throws IOException {
+        if (!f.exists()) {
+            return true;
+        }


Do we have a failing test case for this? Ideally one that covers all filesystems (I'm not sure whether we have such abstract test in place)

Yes, I still have to add that 👍

dmvk · 2021-12-14T14:26:29Z

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

-            } catch (IOException e) {
-                log.warn("Could not properly mark job {} result as clean.", jobId, e);
-            }
+    private void cleanUpJobResult(JobID jobId) {


commitJobResult?

dmvk · 2021-12-14T14:39:14Z

...runtime/src/main/java/org/apache/flink/runtime/jobmaster/CheckpointJobDataCleanupRunner.java

+
+    private void initializeCompletedCheckpointStore() {
+        try {
+            this.completedCheckpointStore =


The cleanup executor uses a fixed thread pool [1], so we're not guaranteed to always have a same thread here.

[1] https://github.com/apache/flink/blob/release-1.14.0/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java#L314

dmvk · 2021-12-14T14:41:13Z

...runtime/src/main/java/org/apache/flink/runtime/jobmaster/CheckpointJobDataCleanupRunner.java

+    public CompletableFuture<Acknowledge> cancel(Time timeout) {
+        Preconditions.checkState(
+                resultFuture != null, "The CheckpointJobDataCleanupRunner was not started, yet.");
+        if (resultFuture.cancel(true)) {


What is the use-case for cancelling here? This will fail the future, but wouldn't the cleanup still keep executing?

You're right. I have to look into it more closely.

dmvk · 2021-12-14T14:42:10Z

...runtime/src/main/java/org/apache/flink/runtime/jobmaster/CheckpointJobDataCleanupRunner.java

+        return new ExecutionGraphInfo(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
+                        jobResult.getJobId(),
+                        "we might want to move the job name into the job result",


do we also need this for the web ui?

yes, that was my understanding. we need to provide some dummy response for now

dmvk · 2021-12-14T16:30:58Z

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java

+    @Override
+    public void localCleanup(JobID jobId) throws IOException {
        checkNotNull(jobId);
+        deleteLocalData(jobId);


Should we also obtain a RW lock here?

Thanks for the pointer, I will verify it.

dmvk · 2021-12-14T16:32:45Z

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java

+        deleteLocalData(jobId);
+    }

+    private void deleteLocalData(JobID jobId) throws IOException {


do we need an extra method? can the globalCleanup simply call the localCleanup instead?

We could do that. I just thought that it makes the reading of the code easier...

dmvk · 2021-12-14T16:52:41Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/ThrowingJobGraphWriter.java

-
-    @Override
-    public void releaseJobGraph(JobID jobId) {}
+    public void localCleanup(JobID jobId) {}


seems unnecessary

dmvk · 2021-12-14T16:55:54Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerUtils.java

            try {
-                return createCompletedCheckpointStore(
-                        configuration, userCodeLoader, checkpointRecoveryFactory, log, jobId);
+                return checkpointRecoveryFactory.createRecoveredCompletedCheckpointStore(


I'm not a big fan of moving this into the CheckpointRecoveryFactory interface. What is the intuition / benefit behind it?

We need to instantiate the CompletedCheckpointStore in different locations (i.e. in SchedulerBase and in CheckpointJobDataCleanupRunner). I didn't want to duplicate the logic for accessing the configuration to get the number of retained checkpoints.

But instead, we could also move the static method from SchedulerUtils into some other utils class

dmvk · 2021-12-14T16:56:58Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobGraphWriter.java

     *
     * @param jobId specifying the job to release the locks for
     * @throws Exception if the locks cannot be released
     */


Do we intentionally keep the old javadoc?

Yes, I thought it would be more specific to the actual class.

…CleanableResource and GloballyCleanableResource

… GloballyCleanableResource

…leResource and GloballyCleanableResource

…esource and GloballyCleanableResource

…overyFactory.createRecoveredCompletedCheckpointStore

…b into more generic ArchivedExecutionGraph.createSparseArchivedExecutionGraph

…ring close

…ate constructor

… JobManager configuration into the interface

…method

…cher

…lobStore interface Previously, FileSystemBlobStore's delete methods didn't return true if the corresponding file didn't exist and, therefore, was not deleted. BlobStore.delete and BlobStore.deleteAll have a different contract which states that, in that case, the method should return true. The internal FileSystemBlobStore.delete method got updated accordingly. Additionally, a test class covering the FileSystemBlobStore was added

XComp · 2022-01-28T07:59:25Z

Closed in favor of #4

XComp mentioned this pull request Dec 10, 2021

[FLINK-11813] Introduces cleanup of dirty jobs in Dispatcher #2

Closed

XComp force-pushed the FLINK-11813-cleanup-v2 branch 2 times, most recently from c91336e to 0dbdeff Compare December 13, 2021 11:43

dmvk reviewed Dec 15, 2021

View reviewed changes

XComp added 21 commits December 15, 2021 16:42

[FLINK-11813] Adds generic interfaces for cleaning up Job-related data

a7d2cee

[FLINK-11813] Refactors JobGraphWriter interface to implement Locally…

70e2394

…CleanableResource and GloballyCleanableResource

[FLINK-11813] Makes BlobServer implement LocallyCleanableResource and…

0ecaba8

… GloballyCleanableResource

[FLINK-11813] Makes HighAvailabilityServices implement LocallyCleanab…

4b7f8ef

…leResource and GloballyCleanableResource

[FLINK-11813] Makes JobManagerMetricGroup implement LocallyCleanableR…

d2fe489

…esource and GloballyCleanableResource

[FLINK-11813] Removes unused classloader parameter from CheckpointRec…

34d754f

…overyFactory.createRecoveredCompletedCheckpointStore

[hotfix] Refactors nested if statements

7fb38c0

[hotfix] Minor fixes in ApplicationStatus

e1a02d7

[FLINK-11813] Renames ArchivedExecutionGraph.createFromInitializingJo…

fc835b1

…b into more generic ArchivedExecutionGraph.createSparseArchivedExecutionGraph

[FLINK-11813] Adds method to make the TestingJobManagerRunner fail du…

704e010

…ring close

[FLINK-11813] Adds JobManagerRunnerRegistry

78e6199

[FLINK-11813] Integrates JobManagerRunnerRegistry into Dispatcher

cb1ca1e

[FLINK-11813] Adds ResourceCleaner interface and default implementation

288bda5

[FLINK-11813] Refactors Dispatcher constructor signature

a76cdda

[FLINK-11813] Moves JobManagerRunnerRegistry instantiation into separ…

19c2d60

…ate constructor

[FLINK-11813] Integrates ResourceCleaner into Dispatcher

b8c34b3

[FLINK-11813] Moves CheckpointRecoveryFactory initialization based on…

e19ffcb

… JobManager configuration into the interface

[FLINK-11813] Adds CheckpointResourcesCleanupRunner

6f7a018

[FLINK-11813] Refactors createJobManagerRunner to be an initializing …

fcb20a8

…method

[FLINK-11813] Integrates CheckpointResourcesCleanupRunner into Dispat…

ba5b891

…cher

[FLINK-11813] Simplifies code

09bc067

XComp force-pushed the FLINK-11813-move-jrs-init branch from 59ac153 to ba31513 Compare December 15, 2021 15:45

XComp force-pushed the FLINK-11813-cleanup-v2 branch from 05155b9 to c513b0d Compare December 21, 2021 16:57

XComp closed this Jan 28, 2022

XComp added a commit that referenced this pull request Oct 21, 2022

commit #3

c969d8b

XComp added a commit that referenced this pull request Mar 22, 2024

foo #3

d77f87f

XComp added a commit that referenced this pull request Apr 2, 2024

fix #3

6d80497

XComp added a commit that referenced this pull request Apr 2, 2024

foo #3

f14e8fd

[FLINK-11813] Introduces basic JobManagerRunner implementation #3

[FLINK-11813] Introduces basic JobManagerRunner implementation #3

Uh oh!

Conversation

XComp commented Dec 9, 2021

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

XComp commented Jan 28, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

XComp commented Jan 28, 2022 •

edited

Loading