[FLINK-25432] Introduces ResourceCleaner that gets triggered for finished jobs that are not cleaned up, yet, according to the JobResultStore

flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherGatewayServiceFactory.java

@@ -26,11 +26,13 @@
 import org.apache.flink.runtime.dispatcher.DispatcherFactory;
 import org.apache.flink.runtime.dispatcher.DispatcherId;
 import org.apache.flink.runtime.dispatcher.PartialDispatcherServices;
-import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobGraphStore;
+import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobPersistenceComponents;
 import org.apache.flink.runtime.dispatcher.runner.AbstractDispatcherLeaderProcess;
 import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherGatewayService;
+import org.apache.flink.runtime.highavailability.JobResultStore;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmanager.JobGraphWriter;
+import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.util.FlinkRuntimeException;
 
@@ -83,7 +85,9 @@ public ApplicationDispatcherGatewayServiceFactory(
     public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
             DispatcherId fencingToken,
             Collection<JobGraph> recoveredJobs,
-            JobGraphWriter jobGraphWriter) {
+            Collection<JobResult> recoveredDirtyJobResults,
+            JobGraphWriter jobGraphWriter,
+            JobResultStore jobResultStore) {
 
         final List<JobID> recoveredJobIds = getRecoveredJobIds(recoveredJobs);
 
@@ -94,6 +98,7 @@ public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
                             rpcService,
                             fencingToken,
                             recoveredJobs,
+                            recoveredDirtyJobResults,
                             (dispatcherGateway, scheduledExecutor, errorHandler) ->
                                     new ApplicationDispatcherBootstrap(
                                             application,
@@ -102,8 +107,8 @@ public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
                                             dispatcherGateway,
                                             scheduledExecutor,
                                             errorHandler),
-                            PartialDispatcherServicesWithJobGraphStore.from(
-                                    partialDispatcherServices, jobGraphWriter));
+                            PartialDispatcherServicesWithJobPersistenceComponents.from(
+                                    partialDispatcherServices, jobGraphWriter, jobResultStore));
         } catch (Exception e) {
             throw new FlinkRuntimeException("Could not create the Dispatcher rpc endpoint.", e);
         }

flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherLeaderProcessFactoryFactory.java

@@ -26,7 +26,7 @@
 import org.apache.flink.runtime.dispatcher.runner.DispatcherLeaderProcessFactory;
 import org.apache.flink.runtime.dispatcher.runner.DispatcherLeaderProcessFactoryFactory;
 import org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcessFactory;
-import org.apache.flink.runtime.jobmanager.JobGraphStoreFactory;
+import org.apache.flink.runtime.jobmanager.JobPersistenceComponentFactory;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
 
@@ -59,7 +59,7 @@ private ApplicationDispatcherLeaderProcessFactoryFactory(
 
     @Override
     public DispatcherLeaderProcessFactory createFactory(
-            JobGraphStoreFactory jobGraphStoreFactory,
+            JobPersistenceComponentFactory jobPersistenceComponentFactory,
             Executor ioExecutor,
             RpcService rpcService,
             PartialDispatcherServices partialDispatcherServices,
@@ -74,7 +74,10 @@ public DispatcherLeaderProcessFactory createFactory(
                         partialDispatcherServices);
 
         return new SessionDispatcherLeaderProcessFactory(
-                dispatcherServiceFactory, jobGraphStoreFactory, ioExecutor, fatalErrorHandler);
+                dispatcherServiceFactory,
+                jobPersistenceComponentFactory,
+                ioExecutor,
+                fatalErrorHandler);
     }
 
     public static ApplicationDispatcherLeaderProcessFactoryFactory create(

flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java

@@ -26,19 +26,25 @@
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.ProgramInvocationException;
 import org.apache.flink.client.testjar.BlockingJob;
+import org.apache.flink.client.testjar.ErrorHandlingSubmissionJob;
 import org.apache.flink.client.testjar.FailingJob;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.configuration.PipelineOptionsInternal;
+import org.apache.flink.core.testutils.FlinkAssertions;
+import org.apache.flink.runtime.client.DuplicateJobSubmissionException;
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
 import org.apache.flink.runtime.dispatcher.SessionDispatcherFactory;
 import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory;
 import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.highavailability.JobResultEntry;
+import org.apache.flink.runtime.highavailability.JobResultStore;
 import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl;
+import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedJobResultStore;
 import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
 import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
@@ -48,11 +54,13 @@
 import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
 import org.apache.flink.runtime.rest.JobRestEndpointFactory;
 import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.TestingJobResultStore;
 import org.apache.flink.testutils.TestingUtils;
 import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
 import java.time.Duration;
 import java.util.UUID;
@@ -61,11 +69,10 @@
 import java.util.function.Supplier;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /** Integration tests related to {@link ApplicationDispatcherBootstrap}. */
-public class ApplicationDispatcherBootstrapITCase extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+public class ApplicationDispatcherBootstrapITCase {
 
     private static final Duration TIMEOUT = Duration.ofMinutes(10);
 
@@ -127,7 +134,9 @@ public void testDispatcherRecoversAfterLosingAndRegainingLeadership() throws Exc
                     cluster.requestJobResult(ApplicationDispatcherBootstrap.ZERO_JOB_ID);
             haServices.revokeDispatcherLeadership();
             // make sure the leadership is revoked to avoid race conditions
-            assertEquals(ApplicationStatus.UNKNOWN, firstJobResult.get().getApplicationStatus());
+            assertThat(firstJobResult.get())
+                    .extracting(JobResult::getApplicationStatus)
+                    .isEqualTo(ApplicationStatus.UNKNOWN);
             haServices.grantDispatcherLeadership();
 
             // job is suspended, wait until it's running
@@ -141,10 +150,11 @@ public void testDispatcherRecoversAfterLosingAndRegainingLeadership() throws Exc
             BlockingJob.unblock(blockId);
 
             // and wait for it to actually finish
-            final CompletableFuture<JobResult> secondJobResult =
-                    cluster.requestJobResult(ApplicationDispatcherBootstrap.ZERO_JOB_ID);
-            assertTrue(secondJobResult.get().isSuccess());
-            assertEquals(ApplicationStatus.SUCCEEDED, secondJobResult.get().getApplicationStatus());
+            final JobResult secondJobResult =
+                    cluster.requestJobResult(ApplicationDispatcherBootstrap.ZERO_JOB_ID).get();
+            assertThat(secondJobResult.isSuccess()).isTrue();
+            assertThat(secondJobResult.getApplicationStatus())
+                    .isEqualTo(ApplicationStatus.SUCCEEDED);
 
             // the cluster should shut down automatically once the application completes
             awaitClusterStopped(cluster, deadline);
@@ -153,6 +163,64 @@ public void testDispatcherRecoversAfterLosingAndRegainingLeadership() throws Exc
         }
     }
 
+    @Test
+    public void testDirtyJobResultRecoveryInApplicationMode() throws Exception {
+        final Deadline deadline = Deadline.fromNow(TIMEOUT);
+        final Configuration configuration = new Configuration();
+        configuration.set(HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());
+        configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
+        configuration.set(ClientOptions.CLIENT_RETRY_PERIOD, Duration.ofMillis(100));
+        final TestingMiniClusterConfiguration clusterConfiguration =
+                TestingMiniClusterConfiguration.newBuilder()
+                        .setConfiguration(configuration)
+                        .build();
+
+        // having a dirty entry in the JobResultStore should make the ApplicationDispatcherBootstrap
+        // implementation fail to submit the job
+        final JobResultStore jobResultStore = new EmbeddedJobResultStore();
+        jobResultStore.createDirtyResult(
+                new JobResultEntry(
+                        TestingJobResultStore.createSuccessfulJobResult(
+                                ApplicationDispatcherBootstrap.ZERO_JOB_ID)));
+        final EmbeddedHaServicesWithLeadershipControl haServices =
+                new EmbeddedHaServicesWithLeadershipControl(TestingUtils.defaultExecutor()) {
+
+                    @Override
+                    public JobResultStore getJobResultStore() {
+                        return jobResultStore;
+                    }
+                };
+
+        final TestingMiniCluster.Builder clusterBuilder =
+                TestingMiniCluster.newBuilder(clusterConfiguration)
+                        .setHighAvailabilityServicesSupplier(() -> haServices)
+                        .setDispatcherResourceManagerComponentFactorySupplier(
+                                createApplicationModeDispatcherResourceManagerComponentFactorySupplier(
+                                        clusterConfiguration.getConfiguration(),
+                                        ErrorHandlingSubmissionJob.createPackagedProgram()));
+        try (final MiniCluster cluster = clusterBuilder.build()) {
+            // start mini cluster and submit the job
+            cluster.start();
+
+            // the cluster should shut down automatically once the application completes
+            awaitClusterStopped(cluster, deadline);
+        }
+
+        FlinkAssertions.assertThatChainOfCauses(ErrorHandlingSubmissionJob.getSubmissionException())
+                .as(
+                        "The job's main method shouldn't have been succeeded due to a DuplicateJobSubmissionException.")
+                .hasAtLeastOneElementOfType(DuplicateJobSubmissionException.class);
+
+        assertThat(
+                        jobResultStore.hasDirtyJobResultEntry(
+                                ApplicationDispatcherBootstrap.ZERO_JOB_ID))
+                .isFalse();
+        assertThat(
+                        jobResultStore.hasCleanJobResultEntry(
+                                ApplicationDispatcherBootstrap.ZERO_JOB_ID))
+                .isTrue();
+    }
+
     @Test
     public void testSubmitFailedJobOnApplicationError() throws Exception {
         final Deadline deadline = Deadline.fromNow(TIMEOUT);

flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapTest.java

@@ -45,14 +45,15 @@
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.TestLoggerExtension;
 import org.apache.flink.util.concurrent.FutureUtils;
 import org.apache.flink.util.concurrent.ScheduledExecutor;
 import org.apache.flink.util.concurrent.ScheduledExecutorServiceAdapter;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -79,7 +80,8 @@
 import static org.junit.jupiter.api.Assertions.fail;
 
 /** Tests for the {@link ApplicationDispatcherBootstrap}. */
-public class ApplicationDispatcherBootstrapTest extends TestLogger {
+@ExtendWith(TestLoggerExtension.class)
+public class ApplicationDispatcherBootstrapTest {
 
     private static final int TIMEOUT_SECONDS = 10;
 
@@ -641,7 +643,7 @@ public void testDuplicateJobSubmissionWithTerminatedJobId() throws Throwable {
      * In this scenario, job result is no longer present in the {@link
      * org.apache.flink.runtime.dispatcher.Dispatcher dispatcher} (job has terminated and job
      * manager failed over), but we know that job has already terminated from {@link
-     * org.apache.flink.runtime.highavailability.RunningJobsRegistry running jobs registry}.
+     * org.apache.flink.runtime.highavailability.JobResultStore}.
      */
     @Test
     public void testDuplicateJobSubmissionWithTerminatedJobIdWithUnknownResult() throws Throwable {
@@ -675,7 +677,7 @@ public void testDuplicateJobSubmissionWithTerminatedJobIdWithUnknownResult() thr
      * In this scenario, job result is no longer present in the {@link
      * org.apache.flink.runtime.dispatcher.Dispatcher dispatcher} (job has terminated and job
      * manager failed over), but we know that job has already terminated from {@link
-     * org.apache.flink.runtime.highavailability.RunningJobsRegistry running jobs registry}.
+     * org.apache.flink.runtime.highavailability.JobResultStore}.
      */
     @Test
     public void testDuplicateJobSubmissionWithTerminatedJobIdWithUnknownResultAttached()

flink-clients/src/test/java/org/apache/flink/client/testjar/ErrorHandlingSubmissionJob.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.client.testjar;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.client.cli.CliFrontendTestUtils;
+import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.util.FlinkException;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.net.MalformedURLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * {@code ErrorHandlingSubmissionJob} provides a factory method for creating a {@link
+ * PackagedProgram} that monitors the job submission within the job's {@code main} method.
+ */
+public class ErrorHandlingSubmissionJob {
+
+    private static final AtomicReference<Exception> SUBMISSION_EXCEPTION = new AtomicReference<>();
+
+    public static PackagedProgram createPackagedProgram() throws FlinkException {
+        try {
+            return PackagedProgram.newBuilder()
+                    .setUserClassPaths(
+                            Collections.singletonList(
+                                    new File(CliFrontendTestUtils.getTestJarPath())
+                                            .toURI()
+                                            .toURL()))
+                    .setEntryPointClassName(ErrorHandlingSubmissionJob.class.getName())
+                    .build();
+        } catch (ProgramInvocationException | FileNotFoundException | MalformedURLException e) {
+            throw new FlinkException("Could not load the provided entrypoint class.", e);
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+        env.fromCollection(Arrays.asList(1, 2, 3))
+                .map(element -> element + 1)
+                .output(new DiscardingOutputFormat<>());
+
+        try {
+            env.execute();
+        } catch (Exception e) {
+            SUBMISSION_EXCEPTION.set(e);
+            throw e;
+        }
+    }
+
+    public static Exception getSubmissionException() {
+        return SUBMISSION_EXCEPTION.get();
+    }
+}