You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "eemario (via GitHub)" <gi...@apache.org> on 2026/03/06 07:50:26 UTC
[PR] [FLINK-38975][runtime] Support application recovery [flink]
eemario opened a new pull request, #27741:
URL: https://github.com/apache/flink/pull/27741
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## 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, 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27741:
URL: https://github.com/apache/flink/pull/27741#issuecomment-4010164123
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "520ebd06b9b9554625e2b5fe3e2819a96488657d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "520ebd06b9b9554625e2b5fe3e2819a96488657d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 520ebd06b9b9554625e2b5fe3e2819a96488657d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973214870
##########
flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.deployment.application;
+
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.configuration.ConfigUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.runtime.application.AbstractApplication;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.blob.PermanentBlobService;
+import org.apache.flink.runtime.jobmanager.ApplicationStoreEntry;
+
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+import java.util.List;
+
+/** {@code ApplicationStoreEntry} for {@code PackagedProgramApplication}. */
+public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
+
+ private final Configuration configuration;
+
+ private final PermanentBlobKey userJarBlobKey;
+
+ private final String entryClass;
+
+ private final String[] programArgs;
+
+ private final ApplicationID applicationId;
+
+ private final String applicationName;
+
+ private final boolean handleFatalError;
+
+ private final boolean enforceSingleJobExecution;
+
+ private final boolean submitFailedJobOnApplicationError;
+
+ private final boolean shutDownOnFinish;
+
+ public PackagedProgramApplicationEntry(
+ Configuration configuration,
+ PermanentBlobKey userJarBlobKey,
+ String entryClass,
+ String[] programArgs,
+ ApplicationID applicationId,
+ String applicationName,
+ boolean handleFatalError,
+ boolean enforceSingleJobExecution,
+ boolean submitFailedJobOnApplicationError,
+ boolean shutDownOnFinish) {
+ this.configuration = configuration;
+ this.userJarBlobKey = userJarBlobKey;
+ this.entryClass = entryClass;
+ this.programArgs = programArgs;
+ this.applicationId = applicationId;
+ this.applicationName = applicationName;
+ this.handleFatalError = handleFatalError;
+ this.enforceSingleJobExecution = enforceSingleJobExecution;
+ this.submitFailedJobOnApplicationError = submitFailedJobOnApplicationError;
+ this.shutDownOnFinish = shutDownOnFinish;
+ }
+
+ @Override
+ public AbstractApplication getApplication(
+ PermanentBlobService blobService,
+ Collection<JobInfo> recoveredJobInfos,
+ Collection<JobInfo> recoveredTerminalJobInfos) {
+ File jarFile;
+ try {
+ jarFile = blobService.getFile(applicationId, userJarBlobKey);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to get user jar file from blob", e);
+ }
+
+ if (!jarFile.exists()) {
+ throw new RuntimeException(String.format("Jar file %s does not exist", jarFile));
+ }
+
+ PackagedProgram program;
+ try {
+ program =
+ PackagedProgram.newBuilder()
+ .setJarFile(jarFile)
+ .setEntryPointClassName(entryClass)
+ .setConfiguration(configuration)
+ .setUserClassPaths(getClasspaths())
+ .setArguments(programArgs)
+ .build();
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to create PackagedProgram for application %s", applicationId));
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2951272433
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultApplicationStore.java:
##########
@@ -0,0 +1,328 @@
+/*
+ * 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.runtime.jobmanager;
+
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.runtime.persistence.ResourceVersion;
+import org.apache.flink.runtime.persistence.StateHandleStore;
+import org.apache.flink.runtime.state.RetrievableStateHandle;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Default implementation for {@link ApplicationStore}. Combined with different {@link
+ * StateHandleStore}, we could persist the applications to various distributed storage.
+ */
+public class DefaultApplicationStore<R extends ResourceVersion<R>> implements ApplicationStore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultApplicationStore.class);
+
+ private final Object lock = new Object();
+
+ /** The set of IDs of all added applications. */
+ @GuardedBy("lock")
+ private final Set<ApplicationID> addedApplications = new HashSet<>();
+
+ /** Submitted applications handle store. */
+ private final StateHandleStore<ApplicationStoreEntry, R> applicationStateHandleStore;
+
+ private final ApplicationStoreUtil applicationStoreUtil;
+
+ /** Flag indicating whether this instance is running. */
+ @GuardedBy("lock")
+ private volatile boolean running;
+
+ public DefaultApplicationStore(
+ StateHandleStore<ApplicationStoreEntry, R> stateHandleStore,
+ ApplicationStoreUtil applicationStoreUtil) {
+ this.applicationStateHandleStore = checkNotNull(stateHandleStore);
+ this.applicationStoreUtil = checkNotNull(applicationStoreUtil);
+
+ this.running = false;
+ }
+
+ @Override
+ public void start() throws Exception {
+ synchronized (lock) {
+ if (!running) {
+ running = true;
+ }
+ }
+ }
+
+ @Override
+ public void stop() throws Exception {
+ synchronized (lock) {
+ if (running) {
+ running = false;
+ LOG.info("Stopping DefaultApplicationStore.");
+ Exception exception = null;
+
+ try {
+ applicationStateHandleStore.releaseAll();
+ } catch (Exception e) {
+ exception = e;
+ }
+
+ if (exception != null) {
+ throw new FlinkException(
+ "Could not properly stop the DefaultApplicationStore.", exception);
+ }
+ }
+ }
+ }
+
+ @Override
+ public Optional<ApplicationStoreEntry> recoverApplication(ApplicationID applicationId)
+ throws Exception {
+ checkNotNull(applicationId, "Application ID");
+
+ LOG.debug("Recovering application {} from {}.", applicationId, applicationStateHandleStore);
+
+ final String name = applicationStoreUtil.applicationIdToName(applicationId);
+
+ synchronized (lock) {
+ verifyIsRunning();
+
+ boolean success = false;
+
+ RetrievableStateHandle<ApplicationStoreEntry> applicationRetrievableStateHandle;
+
+ try {
+ try {
+ applicationRetrievableStateHandle =
+ applicationStateHandleStore.getAndLock(name);
+ } catch (StateHandleStore.NotExistException ignored) {
+ success = true;
+ return Optional.empty();
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not retrieve the submitted application state handle "
+ + "for "
+ + name
+ + " from the submitted application store.",
+ e);
+ }
+
+ ApplicationStoreEntry application;
+ try {
+ application = applicationRetrievableStateHandle.retrieveState();
+ } catch (ClassNotFoundException cnfe) {
+ throw new FlinkException(
+ "Could not retrieve submitted application from state handle under "
+ + name
+ + ". This indicates that you are trying to recover from state written by an "
+ + "older Flink version which is not compatible. Try cleaning the state handle store.",
+ cnfe);
+ } catch (IOException ioe) {
+ throw new FlinkException(
+ "Could not retrieve submitted application from state handle under "
+ + name
+ + ". This indicates that the retrieved state handle is broken. Try cleaning the state handle "
+ + "store.",
+ ioe);
+ }
+
+ addedApplications.add(applicationId);
+
+ LOG.info("Recovered {} ({}).", application.getName(), applicationId);
+
+ success = true;
+ return Optional.of(application);
+ } finally {
+ if (!success) {
+ applicationStateHandleStore.release(name);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void putApplication(ApplicationStoreEntry application) throws Exception {
+ checkNotNull(application, "Application");
+
+ final ApplicationID applicationID = application.getApplicationId();
Review Comment:
applicationID -> applicationId
##########
flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java:
##########
@@ -122,7 +122,8 @@ public AbstractApplication getApplication(
handleFatalError,
enforceSingleJobExecution,
submitFailedJobOnApplicationError,
- shutDownOnFinish);
+ shutDownOnFinish,
+ userJarBlobKey);
Review Comment:
Note that this is not a hotfix but a fixup for this PR.
The change should be squashed into another commit before the PR get merged.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -251,28 +446,40 @@ private Optional<CompletableFuture<Void>> submitAddedJobIfRunning(ExecutionPlan
private CompletableFuture<Void> submitAddedJob(ExecutionPlan executionPlan) {
final DispatcherGateway dispatcherGateway = getDispatcherGatewayInternal();
+ final JobID jobId = executionPlan.getJobID();
+ final Duration timeout =
+ executionPlan.getJobConfiguration().get(RpcOptions.ASK_TIMEOUT_DURATION);
+ // Skip job submission if its associated application exists, as the application will handle
+ // the job recovery
+ ApplicationID applicationId = executionPlan.getApplicationId().orElse(null);
return dispatcherGateway
- .submitJob(executionPlan, RpcUtils.INF_TIMEOUT)
- .thenApply(FunctionUtils.nullFn())
- .exceptionally(this::filterOutDuplicateJobSubmissionException);
- }
-
- private Void filterOutDuplicateJobSubmissionException(Throwable throwable) {
- final Throwable strippedException = ExceptionUtils.stripCompletionException(throwable);
- if (strippedException instanceof DuplicateJobSubmissionException) {
- final DuplicateJobSubmissionException duplicateJobSubmissionException =
- (DuplicateJobSubmissionException) strippedException;
-
- log.debug(
- "Ignore recovered job {} because the job is currently being executed.",
- duplicateJobSubmissionException.getJobID(),
- duplicateJobSubmissionException);
-
- return null;
- } else {
- throw new CompletionException(throwable);
- }
+ .requestApplication(applicationId, timeout)
Review Comment:
Should it avoid calling `requestApplication` if the `applicationId` is null?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java:
##########
@@ -401,6 +402,28 @@ public static ArchivedExecutionGraph createFrom(
executionGraph.getApplicationId().orElse(null));
}
+ /**
+ * Create a {@link ArchivedExecutionGraph} from the given {@link JobResult}.
+ *
+ * @param jobResult to create the ArchivedExecutionGraph from
+ * @param initializationTimestamp optionally overrides the initialization timestamp if the
+ * jobResult does not have a valid one
+ * @return ArchivedExecutionGraph created from the given jobResult
+ */
+ public static ArchivedExecutionGraph createFrom(
Review Comment:
It's better to name it as `createSparseArchivedExecutionGraph` as the result is not a standard `ArchivedExecutionGraph`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973212818
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -251,28 +446,40 @@ private Optional<CompletableFuture<Void>> submitAddedJobIfRunning(ExecutionPlan
private CompletableFuture<Void> submitAddedJob(ExecutionPlan executionPlan) {
final DispatcherGateway dispatcherGateway = getDispatcherGatewayInternal();
+ final JobID jobId = executionPlan.getJobID();
+ final Duration timeout =
+ executionPlan.getJobConfiguration().get(RpcOptions.ASK_TIMEOUT_DURATION);
+ // Skip job submission if its associated application exists, as the application will handle
+ // the job recovery
+ ApplicationID applicationId = executionPlan.getApplicationId().orElse(null);
return dispatcherGateway
- .submitJob(executionPlan, RpcUtils.INF_TIMEOUT)
- .thenApply(FunctionUtils.nullFn())
- .exceptionally(this::filterOutDuplicateJobSubmissionException);
- }
-
- private Void filterOutDuplicateJobSubmissionException(Throwable throwable) {
- final Throwable strippedException = ExceptionUtils.stripCompletionException(throwable);
- if (strippedException instanceof DuplicateJobSubmissionException) {
- final DuplicateJobSubmissionException duplicateJobSubmissionException =
- (DuplicateJobSubmissionException) strippedException;
-
- log.debug(
- "Ignore recovered job {} because the job is currently being executed.",
- duplicateJobSubmissionException.getJobID(),
- duplicateJobSubmissionException);
-
- return null;
- } else {
- throw new CompletionException(throwable);
- }
+ .requestApplication(applicationId, timeout)
Review Comment:
Updated: throws early if applicationId is null, as this indicates an abnormal state that should trigger a fatal error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk merged PR #27741:
URL: https://github.com/apache/flink/pull/27741
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973219496
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -98,46 +123,93 @@ private void startServices() {
getClass().getSimpleName()),
e);
}
+
+ try {
+ applicationStore.start();
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Could not start %s when trying to start the %s.",
+ applicationStore.getClass().getSimpleName(),
+ getClass().getSimpleName()),
+ e);
+ }
}
private void createDispatcherIfRunning(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
runIfStateIs(
- State.RUNNING, () -> createDispatcher(executionPlans, recoveredDirtyJobResults));
+ State.RUNNING,
+ () ->
+ createDispatcher(
+ executionPlans,
+ recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults));
}
private void createDispatcher(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
final DispatcherGatewayService dispatcherService =
dispatcherGatewayServiceFactory.create(
DispatcherId.fromUuid(getLeaderSessionId()),
executionPlans,
recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults,
executionPlanStore,
- jobResultStore);
+ jobResultStore,
+ applicationStore,
+ applicationResultStore);
completeDispatcherSetup(dispatcherService);
}
- private CompletableFuture<Void>
- createDispatcherBasedOnRecoveredExecutionPlansAndRecoveredDirtyJobResults() {
- // TODO support application recovery which may require fetching user jar from blob server
-
- final CompletableFuture<Collection<JobResult>> dirtyJobsFuture =
+ private CompletableFuture<Void> createDispatcherBasedOnRecoveredApplicationsAndJobs() {
+ final CompletableFuture<Collection<JobResult>> dirtyJobResultsFuture =
CompletableFuture.supplyAsync(this::getDirtyJobResultsIfRunning, ioExecutor);
- return dirtyJobsFuture
- .thenApplyAsync(
- dirtyJobs ->
- this.recoverJobsIfRunning(
- dirtyJobs.stream()
+ final CompletableFuture<Collection<ExecutionPlan>> recoveredJobsFuture =
+ dirtyJobResultsFuture.thenApplyAsync(
+ dirtyJobResults ->
+ recoverJobsIfRunning(
+ dirtyJobResults.stream()
.map(JobResult::getJobId)
.collect(Collectors.toSet())),
- ioExecutor)
- .thenAcceptBoth(dirtyJobsFuture, this::createDispatcherIfRunning)
+ ioExecutor);
+
+ final CompletableFuture<Collection<ApplicationResult>> dirtyApplicationResultsFuture =
+ CompletableFuture.supplyAsync(
+ this::getDirtyApplicationResultsIfRunning, ioExecutor);
+
+ final CompletableFuture<Collection<AbstractApplication>> recoveredApplicationsFuture =
+ dirtyApplicationResultsFuture.thenCombineAsync(
+ recoveredJobsFuture,
+ (dirtyApplicationResults, recoveredJobs) -> {
+ return recoverApplicationsIfRunning(
+ dirtyApplicationResults.stream()
+ .map(ApplicationResult::getApplicationId)
+ .collect(Collectors.toSet()),
+ recoveredJobsFuture.join(),
Review Comment:
Updated: use `allOf` instead.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -887,6 +1062,83 @@ private void writeToArchivedApplicationStore(ArchivedApplication archivedApplica
}
}
+ private CompletableFuture<?> registerGloballyTerminatedApplicationInApplicationResultStore(
+ ArchivedApplication application) {
+ final ApplicationID applicationId = application.getApplicationId();
+
+ return applicationResultStore
+ .hasCleanApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasCleanResult -> {
+ if (hasCleanResult) {
+ log.warn(
+ "Application {} is already marked as clean but clean up was triggered again.",
+ applicationId);
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .hasDirtyApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasDirtyResult -> {
+ if (hasDirtyResult) {
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .createDirtyResultAsync(
+ new ApplicationResultEntry(
+ ApplicationResult
+ .createFrom(
+ application)));
+ });
+ })
+ .handleAsync(
+ (ignored, error) -> {
+ if (error != null) {
+ fatalErrorHandler.onFatalError(
+ new FlinkException(
+ String.format(
+ "The application %s couldn't be marked as pre-cleanup finished in ApplicationResultStore.",
+ applicationId),
+ error));
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973223279
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -1051,7 +1302,9 @@ private void runJob(
final JobID jobId = jobManagerRunner.getJobID();
- partialExecutionGraphInfoStore.put(jobId, new CompletableFuture<>());
+ jobCreateDirtyResultFutures.put(jobId, new CompletableFuture<>());
+ jobMarkResultCleanFutures.put(jobId, new CompletableFuture<>());
+ jobIdsToApplicationIds.put(jobId, applicationId);
Review Comment:
Updated: remove the items after they are not needed anymore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973222058
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -887,6 +1062,83 @@ private void writeToArchivedApplicationStore(ArchivedApplication archivedApplica
}
}
+ private CompletableFuture<?> registerGloballyTerminatedApplicationInApplicationResultStore(
+ ArchivedApplication application) {
+ final ApplicationID applicationId = application.getApplicationId();
+
+ return applicationResultStore
+ .hasCleanApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasCleanResult -> {
+ if (hasCleanResult) {
+ log.warn(
+ "Application {} is already marked as clean but clean up was triggered again.",
+ applicationId);
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .hasDirtyApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasDirtyResult -> {
+ if (hasDirtyResult) {
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .createDirtyResultAsync(
+ new ApplicationResultEntry(
+ ApplicationResult
+ .createFrom(
+ application)));
+ });
+ })
+ .handleAsync(
+ (ignored, error) -> {
+ if (error != null) {
+ fatalErrorHandler.onFatalError(
+ new FlinkException(
+ String.format(
+ "The application %s couldn't be marked as pre-cleanup finished in ApplicationResultStore.",
+ applicationId),
+ error));
+ }
+ applicationCreateDirtyResultFutures.get(applicationId).complete(null);
+ return null;
+ },
+ getMainThreadExecutor());
+ }
+
+ private CompletableFuture<Void> removeApplication(
+ ApplicationID applicationId, Collection<JobID> jobs) {
+ return applicationResourceCleaner
+ .cleanupAsync(applicationId)
+ .thenCombine(
+ FutureUtils.waitForAll(
+ jobs.stream()
+ .map(jobMarkResultCleanFutures::get)
+ .collect(Collectors.toList())),
+ (unused1, unused2) ->
+ applicationResultStore.markResultAsCleanAsync(applicationId))
Review Comment:
Updated: Use `thenCompose` instead of `thenCombine` to properly handle futures that complete exceptionally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973216177
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/DefaultApplicationResourceCleaner.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.runtime.dispatcher.cleanup;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.concurrent.RetryStrategy;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+
+/**
+ * {@code DefaultApplicationResourceCleaner} is the default implementation of {@link
+ * ApplicationResourceCleaner}. It will try to clean up any resource that was added. Failure will
+ * result in an individual retry of the cleanup. The overall cleanup result succeeds after all
+ * subtasks succeeded.
+ */
+public class DefaultApplicationResourceCleaner<T> implements ApplicationResourceCleaner {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DefaultApplicationResourceCleaner.class);
+
+ private final ComponentMainThreadExecutor mainThreadExecutor;
+ private final Executor cleanupExecutor;
+ private final CleanupFn<T> cleanupFn;
+
+ private final Collection<CleanupWithLabel<T>> regularCleanup;
+
+ private final RetryStrategy retryStrategy;
+
+ public static Builder<GloballyCleanableApplicationResource> forGloballyCleanableResources(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ RetryStrategy retryStrategy) {
+ return forCleanableResources(
+ mainThreadExecutor,
+ cleanupExecutor,
+ GloballyCleanableApplicationResource::globalCleanupAsync,
+ retryStrategy);
+ }
+
+ @VisibleForTesting
+ static <T> Builder<T> forCleanableResources(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFunction,
+ RetryStrategy retryStrategy) {
+ return new Builder<>(mainThreadExecutor, cleanupExecutor, cleanupFunction, retryStrategy);
+ }
+
+ @VisibleForTesting
+ @FunctionalInterface
+ interface CleanupFn<T> {
+ CompletableFuture<Void> cleanupAsync(
+ T resource, ApplicationID applicationId, Executor cleanupExecutor);
+ }
+
+ /**
+ * {@code Builder} for creating {@code DefaultApplicationResourceCleaner} instances.
+ *
+ * @param <T> The functional interface that's being translated into the internally used {@link
+ * CleanupFn}.
+ */
+ public static class Builder<T> {
+
+ private final ComponentMainThreadExecutor mainThreadExecutor;
+ private final Executor cleanupExecutor;
+ private final CleanupFn<T> cleanupFn;
+
+ private final RetryStrategy retryStrategy;
+
+ private final Collection<CleanupWithLabel<T>> prioritizedCleanup = new ArrayList<>();
+ private final Collection<CleanupWithLabel<T>> regularCleanup = new ArrayList<>();
+
+ private Builder(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFn,
+ RetryStrategy retryStrategy) {
+ this.mainThreadExecutor = mainThreadExecutor;
+ this.cleanupExecutor = cleanupExecutor;
+ this.cleanupFn = cleanupFn;
+ this.retryStrategy = retryStrategy;
+ }
+
+ /**
+ * Prioritized cleanups run before their regular counterparts. This method enables the
+ * caller to model dependencies between cleanup tasks. The order in which cleanable
+ * resources are added matters, i.e. if two cleanable resources are added as prioritized
+ * cleanup tasks, the resource being added first will block the cleanup of the second
+ * resource. All prioritized cleanup resources will run and finish before any resource that
+ * is added using {@link #withRegularCleanup(String, Object)} is started.
+ *
+ * @param label The label being used when logging errors in the given cleanup.
+ * @param prioritizedCleanup The cleanup callback that is going to be prioritized.
+ */
+ public Builder<T> withPrioritizedCleanup(String label, T prioritizedCleanup) {
+ this.prioritizedCleanup.add(new CleanupWithLabel<>(prioritizedCleanup, label));
+ return this;
+ }
+
+ /**
+ * Regular cleanups are resources for which the cleanup is triggered after all prioritized
+ * cleanups succeeded. All added regular cleanups will run concurrently to each other.
+ *
+ * @param label The label being used when logging errors in the given cleanup.
+ * @param regularCleanup The cleanup callback that is going to run after all prioritized
+ * cleanups are finished.
+ * @see #withPrioritizedCleanup(String, Object)
+ */
+ public Builder<T> withRegularCleanup(String label, T regularCleanup) {
+ this.regularCleanup.add(new CleanupWithLabel<>(regularCleanup, label));
+ return this;
+ }
+
+ public DefaultApplicationResourceCleaner build() {
+ return new DefaultApplicationResourceCleaner<>(
+ mainThreadExecutor,
+ cleanupExecutor,
+ cleanupFn,
+ prioritizedCleanup,
+ regularCleanup,
+ retryStrategy);
+ }
+ }
+
+ private DefaultApplicationResourceCleaner(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFn,
+ Collection<CleanupWithLabel<T>> prioritizedCleanup,
+ Collection<CleanupWithLabel<T>> regularCleanup,
+ RetryStrategy retryStrategy) {
+ this.mainThreadExecutor = mainThreadExecutor;
+ this.cleanupExecutor = cleanupExecutor;
+ this.cleanupFn = cleanupFn;
+ this.regularCleanup = regularCleanup;
+ this.retryStrategy = retryStrategy;
Review Comment:
Updated: removed prioritizedCleanup since it is unnecessary for application resources.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973214149
##########
flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java:
##########
@@ -122,7 +122,8 @@ public AbstractApplication getApplication(
handleFatalError,
enforceSingleJobExecution,
submitFailedJobOnApplicationError,
- shutDownOnFinish);
+ shutDownOnFinish,
+ userJarBlobKey);
Review Comment:
Updated: mark this commit for squashing.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -428,24 +484,53 @@ public void onStart() throws Exception {
this::onFatalError);
if (dispatcherBootstrap instanceof ApplicationBootstrap) {
- // defer starting recovered jobs, as they might be skipped based on user logic
- internalSubmitApplication(((ApplicationBootstrap) dispatcherBootstrap).getApplication())
- .get();
+ // Application Mode
+ checkState(suspendedApplications.isEmpty());
+ checkState(recoveredDirtyApplicationResults.size() <= 1);
+
+ AbstractApplication application =
+ ((ApplicationBootstrap) dispatcherBootstrap).getApplication();
+ if (!recoveredDirtyApplicationResults.isEmpty()) {
+ // the application is already terminated
+ ApplicationResult applicationResult =
+ recoveredDirtyApplicationResults.iterator().next();
+ checkState(application.getApplicationId() == applicationResult.getApplicationId());
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2959488533
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -428,24 +484,53 @@ public void onStart() throws Exception {
this::onFatalError);
if (dispatcherBootstrap instanceof ApplicationBootstrap) {
- // defer starting recovered jobs, as they might be skipped based on user logic
- internalSubmitApplication(((ApplicationBootstrap) dispatcherBootstrap).getApplication())
- .get();
+ // Application Mode
+ checkState(suspendedApplications.isEmpty());
+ checkState(recoveredDirtyApplicationResults.size() <= 1);
+
+ AbstractApplication application =
+ ((ApplicationBootstrap) dispatcherBootstrap).getApplication();
+ if (!recoveredDirtyApplicationResults.isEmpty()) {
+ // the application is already terminated
+ ApplicationResult applicationResult =
+ recoveredDirtyApplicationResults.iterator().next();
+ checkState(application.getApplicationId() == applicationResult.getApplicationId());
Review Comment:
== -> equals
##########
flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplicationEntry.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.deployment.application;
+
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.configuration.ConfigUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.runtime.application.AbstractApplication;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.blob.PermanentBlobService;
+import org.apache.flink.runtime.jobmanager.ApplicationStoreEntry;
+
+import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collection;
+import java.util.List;
+
+/** {@code ApplicationStoreEntry} for {@code PackagedProgramApplication}. */
+public class PackagedProgramApplicationEntry implements ApplicationStoreEntry {
+
+ private final Configuration configuration;
+
+ private final PermanentBlobKey userJarBlobKey;
+
+ private final String entryClass;
+
+ private final String[] programArgs;
+
+ private final ApplicationID applicationId;
+
+ private final String applicationName;
+
+ private final boolean handleFatalError;
+
+ private final boolean enforceSingleJobExecution;
+
+ private final boolean submitFailedJobOnApplicationError;
+
+ private final boolean shutDownOnFinish;
+
+ public PackagedProgramApplicationEntry(
+ Configuration configuration,
+ PermanentBlobKey userJarBlobKey,
+ String entryClass,
+ String[] programArgs,
+ ApplicationID applicationId,
+ String applicationName,
+ boolean handleFatalError,
+ boolean enforceSingleJobExecution,
+ boolean submitFailedJobOnApplicationError,
+ boolean shutDownOnFinish) {
+ this.configuration = configuration;
+ this.userJarBlobKey = userJarBlobKey;
+ this.entryClass = entryClass;
+ this.programArgs = programArgs;
+ this.applicationId = applicationId;
+ this.applicationName = applicationName;
+ this.handleFatalError = handleFatalError;
+ this.enforceSingleJobExecution = enforceSingleJobExecution;
+ this.submitFailedJobOnApplicationError = submitFailedJobOnApplicationError;
+ this.shutDownOnFinish = shutDownOnFinish;
+ }
+
+ @Override
+ public AbstractApplication getApplication(
+ PermanentBlobService blobService,
+ Collection<JobInfo> recoveredJobInfos,
+ Collection<JobInfo> recoveredTerminalJobInfos) {
+ File jarFile;
+ try {
+ jarFile = blobService.getFile(applicationId, userJarBlobKey);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to get user jar file from blob", e);
+ }
+
+ if (!jarFile.exists()) {
+ throw new RuntimeException(String.format("Jar file %s does not exist", jarFile));
+ }
+
+ PackagedProgram program;
+ try {
+ program =
+ PackagedProgram.newBuilder()
+ .setJarFile(jarFile)
+ .setEntryPointClassName(entryClass)
+ .setConfiguration(configuration)
+ .setUserClassPaths(getClasspaths())
+ .setArguments(programArgs)
+ .build();
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to create PackagedProgram for application %s", applicationId));
Review Comment:
The root cause should be included.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -887,6 +1062,83 @@ private void writeToArchivedApplicationStore(ArchivedApplication archivedApplica
}
}
+ private CompletableFuture<?> registerGloballyTerminatedApplicationInApplicationResultStore(
+ ArchivedApplication application) {
+ final ApplicationID applicationId = application.getApplicationId();
+
+ return applicationResultStore
+ .hasCleanApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasCleanResult -> {
+ if (hasCleanResult) {
+ log.warn(
+ "Application {} is already marked as clean but clean up was triggered again.",
+ applicationId);
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .hasDirtyApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasDirtyResult -> {
+ if (hasDirtyResult) {
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .createDirtyResultAsync(
+ new ApplicationResultEntry(
+ ApplicationResult
+ .createFrom(
+ application)));
+ });
+ })
+ .handleAsync(
+ (ignored, error) -> {
+ if (error != null) {
+ fatalErrorHandler.onFatalError(
+ new FlinkException(
+ String.format(
+ "The application %s couldn't be marked as pre-cleanup finished in ApplicationResultStore.",
+ applicationId),
+ error));
Review Comment:
It should return instead of continuing.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -887,6 +1062,83 @@ private void writeToArchivedApplicationStore(ArchivedApplication archivedApplica
}
}
+ private CompletableFuture<?> registerGloballyTerminatedApplicationInApplicationResultStore(
+ ArchivedApplication application) {
+ final ApplicationID applicationId = application.getApplicationId();
+
+ return applicationResultStore
+ .hasCleanApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasCleanResult -> {
+ if (hasCleanResult) {
+ log.warn(
+ "Application {} is already marked as clean but clean up was triggered again.",
+ applicationId);
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .hasDirtyApplicationResultEntryAsync(applicationId)
+ .thenCompose(
+ hasDirtyResult -> {
+ if (hasDirtyResult) {
+ return FutureUtils.completedVoidFuture();
+ }
+
+ return applicationResultStore
+ .createDirtyResultAsync(
+ new ApplicationResultEntry(
+ ApplicationResult
+ .createFrom(
+ application)));
+ });
+ })
+ .handleAsync(
+ (ignored, error) -> {
+ if (error != null) {
+ fatalErrorHandler.onFatalError(
+ new FlinkException(
+ String.format(
+ "The application %s couldn't be marked as pre-cleanup finished in ApplicationResultStore.",
+ applicationId),
+ error));
+ }
+ applicationCreateDirtyResultFutures.get(applicationId).complete(null);
+ return null;
+ },
+ getMainThreadExecutor());
+ }
+
+ private CompletableFuture<Void> removeApplication(
+ ApplicationID applicationId, Collection<JobID> jobs) {
+ return applicationResourceCleaner
+ .cleanupAsync(applicationId)
+ .thenCombine(
+ FutureUtils.waitForAll(
+ jobs.stream()
+ .map(jobMarkResultCleanFutures::get)
+ .collect(Collectors.toList())),
+ (unused1, unused2) ->
+ applicationResultStore.markResultAsCleanAsync(applicationId))
Review Comment:
If `markResultAsCleanAsync` did not throw an exception directly, but instead returned a future that is completed exceptionally, the exceptional future will be ignored.
The same problem also applies to the `markResultAsCleanAsync` of jobs.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -98,46 +123,93 @@ private void startServices() {
getClass().getSimpleName()),
e);
}
+
+ try {
+ applicationStore.start();
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Could not start %s when trying to start the %s.",
+ applicationStore.getClass().getSimpleName(),
+ getClass().getSimpleName()),
+ e);
+ }
}
private void createDispatcherIfRunning(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
runIfStateIs(
- State.RUNNING, () -> createDispatcher(executionPlans, recoveredDirtyJobResults));
+ State.RUNNING,
+ () ->
+ createDispatcher(
+ executionPlans,
+ recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults));
}
private void createDispatcher(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
final DispatcherGatewayService dispatcherService =
dispatcherGatewayServiceFactory.create(
DispatcherId.fromUuid(getLeaderSessionId()),
executionPlans,
recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults,
executionPlanStore,
- jobResultStore);
+ jobResultStore,
+ applicationStore,
+ applicationResultStore);
completeDispatcherSetup(dispatcherService);
}
- private CompletableFuture<Void>
- createDispatcherBasedOnRecoveredExecutionPlansAndRecoveredDirtyJobResults() {
- // TODO support application recovery which may require fetching user jar from blob server
-
- final CompletableFuture<Collection<JobResult>> dirtyJobsFuture =
+ private CompletableFuture<Void> createDispatcherBasedOnRecoveredApplicationsAndJobs() {
+ final CompletableFuture<Collection<JobResult>> dirtyJobResultsFuture =
CompletableFuture.supplyAsync(this::getDirtyJobResultsIfRunning, ioExecutor);
- return dirtyJobsFuture
- .thenApplyAsync(
- dirtyJobs ->
- this.recoverJobsIfRunning(
- dirtyJobs.stream()
+ final CompletableFuture<Collection<ExecutionPlan>> recoveredJobsFuture =
+ dirtyJobResultsFuture.thenApplyAsync(
+ dirtyJobResults ->
+ recoverJobsIfRunning(
+ dirtyJobResults.stream()
.map(JobResult::getJobId)
.collect(Collectors.toSet())),
- ioExecutor)
- .thenAcceptBoth(dirtyJobsFuture, this::createDispatcherIfRunning)
+ ioExecutor);
+
+ final CompletableFuture<Collection<ApplicationResult>> dirtyApplicationResultsFuture =
+ CompletableFuture.supplyAsync(
+ this::getDirtyApplicationResultsIfRunning, ioExecutor);
+
+ final CompletableFuture<Collection<AbstractApplication>> recoveredApplicationsFuture =
+ dirtyApplicationResultsFuture.thenCombineAsync(
+ recoveredJobsFuture,
+ (dirtyApplicationResults, recoveredJobs) -> {
+ return recoverApplicationsIfRunning(
+ dirtyApplicationResults.stream()
+ .map(ApplicationResult::getApplicationId)
+ .collect(Collectors.toSet()),
+ recoveredJobsFuture.join(),
Review Comment:
`recoveredJobs` is already available.
And it's better to use `thenAcceptBoth / allOf` instead of `join`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -1051,7 +1302,9 @@ private void runJob(
final JobID jobId = jobManagerRunner.getJobID();
- partialExecutionGraphInfoStore.put(jobId, new CompletableFuture<>());
+ jobCreateDirtyResultFutures.put(jobId, new CompletableFuture<>());
+ jobMarkResultCleanFutures.put(jobId, new CompletableFuture<>());
+ jobIdsToApplicationIds.put(jobId, applicationId);
Review Comment:
The items added to these three maps are never removed. It will lead to memory leak in session mode.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/DefaultApplicationResourceCleaner.java:
##########
@@ -0,0 +1,237 @@
+/*
+ * 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.runtime.dispatcher.cleanup;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.concurrent.RetryStrategy;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+
+/**
+ * {@code DefaultApplicationResourceCleaner} is the default implementation of {@link
+ * ApplicationResourceCleaner}. It will try to clean up any resource that was added. Failure will
+ * result in an individual retry of the cleanup. The overall cleanup result succeeds after all
+ * subtasks succeeded.
+ */
+public class DefaultApplicationResourceCleaner<T> implements ApplicationResourceCleaner {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DefaultApplicationResourceCleaner.class);
+
+ private final ComponentMainThreadExecutor mainThreadExecutor;
+ private final Executor cleanupExecutor;
+ private final CleanupFn<T> cleanupFn;
+
+ private final Collection<CleanupWithLabel<T>> regularCleanup;
+
+ private final RetryStrategy retryStrategy;
+
+ public static Builder<GloballyCleanableApplicationResource> forGloballyCleanableResources(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ RetryStrategy retryStrategy) {
+ return forCleanableResources(
+ mainThreadExecutor,
+ cleanupExecutor,
+ GloballyCleanableApplicationResource::globalCleanupAsync,
+ retryStrategy);
+ }
+
+ @VisibleForTesting
+ static <T> Builder<T> forCleanableResources(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFunction,
+ RetryStrategy retryStrategy) {
+ return new Builder<>(mainThreadExecutor, cleanupExecutor, cleanupFunction, retryStrategy);
+ }
+
+ @VisibleForTesting
+ @FunctionalInterface
+ interface CleanupFn<T> {
+ CompletableFuture<Void> cleanupAsync(
+ T resource, ApplicationID applicationId, Executor cleanupExecutor);
+ }
+
+ /**
+ * {@code Builder} for creating {@code DefaultApplicationResourceCleaner} instances.
+ *
+ * @param <T> The functional interface that's being translated into the internally used {@link
+ * CleanupFn}.
+ */
+ public static class Builder<T> {
+
+ private final ComponentMainThreadExecutor mainThreadExecutor;
+ private final Executor cleanupExecutor;
+ private final CleanupFn<T> cleanupFn;
+
+ private final RetryStrategy retryStrategy;
+
+ private final Collection<CleanupWithLabel<T>> prioritizedCleanup = new ArrayList<>();
+ private final Collection<CleanupWithLabel<T>> regularCleanup = new ArrayList<>();
+
+ private Builder(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFn,
+ RetryStrategy retryStrategy) {
+ this.mainThreadExecutor = mainThreadExecutor;
+ this.cleanupExecutor = cleanupExecutor;
+ this.cleanupFn = cleanupFn;
+ this.retryStrategy = retryStrategy;
+ }
+
+ /**
+ * Prioritized cleanups run before their regular counterparts. This method enables the
+ * caller to model dependencies between cleanup tasks. The order in which cleanable
+ * resources are added matters, i.e. if two cleanable resources are added as prioritized
+ * cleanup tasks, the resource being added first will block the cleanup of the second
+ * resource. All prioritized cleanup resources will run and finish before any resource that
+ * is added using {@link #withRegularCleanup(String, Object)} is started.
+ *
+ * @param label The label being used when logging errors in the given cleanup.
+ * @param prioritizedCleanup The cleanup callback that is going to be prioritized.
+ */
+ public Builder<T> withPrioritizedCleanup(String label, T prioritizedCleanup) {
+ this.prioritizedCleanup.add(new CleanupWithLabel<>(prioritizedCleanup, label));
+ return this;
+ }
+
+ /**
+ * Regular cleanups are resources for which the cleanup is triggered after all prioritized
+ * cleanups succeeded. All added regular cleanups will run concurrently to each other.
+ *
+ * @param label The label being used when logging errors in the given cleanup.
+ * @param regularCleanup The cleanup callback that is going to run after all prioritized
+ * cleanups are finished.
+ * @see #withPrioritizedCleanup(String, Object)
+ */
+ public Builder<T> withRegularCleanup(String label, T regularCleanup) {
+ this.regularCleanup.add(new CleanupWithLabel<>(regularCleanup, label));
+ return this;
+ }
+
+ public DefaultApplicationResourceCleaner build() {
+ return new DefaultApplicationResourceCleaner<>(
+ mainThreadExecutor,
+ cleanupExecutor,
+ cleanupFn,
+ prioritizedCleanup,
+ regularCleanup,
+ retryStrategy);
+ }
+ }
+
+ private DefaultApplicationResourceCleaner(
+ ComponentMainThreadExecutor mainThreadExecutor,
+ Executor cleanupExecutor,
+ CleanupFn<T> cleanupFn,
+ Collection<CleanupWithLabel<T>> prioritizedCleanup,
+ Collection<CleanupWithLabel<T>> regularCleanup,
+ RetryStrategy retryStrategy) {
+ this.mainThreadExecutor = mainThreadExecutor;
+ this.cleanupExecutor = cleanupExecutor;
+ this.cleanupFn = cleanupFn;
+ this.regularCleanup = regularCleanup;
+ this.retryStrategy = retryStrategy;
Review Comment:
The prioritizedCleanup is not recorded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973210441
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultApplicationStore.java:
##########
@@ -0,0 +1,328 @@
+/*
+ * 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.runtime.jobmanager;
+
+import org.apache.flink.api.common.ApplicationID;
+import org.apache.flink.runtime.persistence.ResourceVersion;
+import org.apache.flink.runtime.persistence.StateHandleStore;
+import org.apache.flink.runtime.state.RetrievableStateHandle;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Default implementation for {@link ApplicationStore}. Combined with different {@link
+ * StateHandleStore}, we could persist the applications to various distributed storage.
+ */
+public class DefaultApplicationStore<R extends ResourceVersion<R>> implements ApplicationStore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultApplicationStore.class);
+
+ private final Object lock = new Object();
+
+ /** The set of IDs of all added applications. */
+ @GuardedBy("lock")
+ private final Set<ApplicationID> addedApplications = new HashSet<>();
+
+ /** Submitted applications handle store. */
+ private final StateHandleStore<ApplicationStoreEntry, R> applicationStateHandleStore;
+
+ private final ApplicationStoreUtil applicationStoreUtil;
+
+ /** Flag indicating whether this instance is running. */
+ @GuardedBy("lock")
+ private volatile boolean running;
+
+ public DefaultApplicationStore(
+ StateHandleStore<ApplicationStoreEntry, R> stateHandleStore,
+ ApplicationStoreUtil applicationStoreUtil) {
+ this.applicationStateHandleStore = checkNotNull(stateHandleStore);
+ this.applicationStoreUtil = checkNotNull(applicationStoreUtil);
+
+ this.running = false;
+ }
+
+ @Override
+ public void start() throws Exception {
+ synchronized (lock) {
+ if (!running) {
+ running = true;
+ }
+ }
+ }
+
+ @Override
+ public void stop() throws Exception {
+ synchronized (lock) {
+ if (running) {
+ running = false;
+ LOG.info("Stopping DefaultApplicationStore.");
+ Exception exception = null;
+
+ try {
+ applicationStateHandleStore.releaseAll();
+ } catch (Exception e) {
+ exception = e;
+ }
+
+ if (exception != null) {
+ throw new FlinkException(
+ "Could not properly stop the DefaultApplicationStore.", exception);
+ }
+ }
+ }
+ }
+
+ @Override
+ public Optional<ApplicationStoreEntry> recoverApplication(ApplicationID applicationId)
+ throws Exception {
+ checkNotNull(applicationId, "Application ID");
+
+ LOG.debug("Recovering application {} from {}.", applicationId, applicationStateHandleStore);
+
+ final String name = applicationStoreUtil.applicationIdToName(applicationId);
+
+ synchronized (lock) {
+ verifyIsRunning();
+
+ boolean success = false;
+
+ RetrievableStateHandle<ApplicationStoreEntry> applicationRetrievableStateHandle;
+
+ try {
+ try {
+ applicationRetrievableStateHandle =
+ applicationStateHandleStore.getAndLock(name);
+ } catch (StateHandleStore.NotExistException ignored) {
+ success = true;
+ return Optional.empty();
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not retrieve the submitted application state handle "
+ + "for "
+ + name
+ + " from the submitted application store.",
+ e);
+ }
+
+ ApplicationStoreEntry application;
+ try {
+ application = applicationRetrievableStateHandle.retrieveState();
+ } catch (ClassNotFoundException cnfe) {
+ throw new FlinkException(
+ "Could not retrieve submitted application from state handle under "
+ + name
+ + ". This indicates that you are trying to recover from state written by an "
+ + "older Flink version which is not compatible. Try cleaning the state handle store.",
+ cnfe);
+ } catch (IOException ioe) {
+ throw new FlinkException(
+ "Could not retrieve submitted application from state handle under "
+ + name
+ + ". This indicates that the retrieved state handle is broken. Try cleaning the state handle "
+ + "store.",
+ ioe);
+ }
+
+ addedApplications.add(applicationId);
+
+ LOG.info("Recovered {} ({}).", application.getName(), applicationId);
+
+ success = true;
+ return Optional.of(application);
+ } finally {
+ if (!success) {
+ applicationStateHandleStore.release(name);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void putApplication(ApplicationStoreEntry application) throws Exception {
+ checkNotNull(application, "Application");
+
+ final ApplicationID applicationID = application.getApplicationId();
Review Comment:
Updated.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java:
##########
@@ -401,6 +402,28 @@ public static ArchivedExecutionGraph createFrom(
executionGraph.getApplicationId().orElse(null));
}
+ /**
+ * Create a {@link ArchivedExecutionGraph} from the given {@link JobResult}.
+ *
+ * @param jobResult to create the ArchivedExecutionGraph from
+ * @param initializationTimestamp optionally overrides the initialization timestamp if the
+ * jobResult does not have a valid one
+ * @return ArchivedExecutionGraph created from the given jobResult
+ */
+ public static ArchivedExecutionGraph createFrom(
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38975][runtime] Support application recovery [flink]
Posted by "eemario (via GitHub)" <gi...@apache.org>.
eemario commented on code in PR #27741:
URL: https://github.com/apache/flink/pull/27741#discussion_r2973219496
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -98,46 +123,93 @@ private void startServices() {
getClass().getSimpleName()),
e);
}
+
+ try {
+ applicationStore.start();
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "Could not start %s when trying to start the %s.",
+ applicationStore.getClass().getSimpleName(),
+ getClass().getSimpleName()),
+ e);
+ }
}
private void createDispatcherIfRunning(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
runIfStateIs(
- State.RUNNING, () -> createDispatcher(executionPlans, recoveredDirtyJobResults));
+ State.RUNNING,
+ () ->
+ createDispatcher(
+ executionPlans,
+ recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults));
}
private void createDispatcher(
Collection<ExecutionPlan> executionPlans,
- Collection<JobResult> recoveredDirtyJobResults) {
+ Collection<JobResult> recoveredDirtyJobResults,
+ Collection<AbstractApplication> recoveredApplications,
+ Collection<ApplicationResult> recoveredDirtyApplicationResults) {
final DispatcherGatewayService dispatcherService =
dispatcherGatewayServiceFactory.create(
DispatcherId.fromUuid(getLeaderSessionId()),
executionPlans,
recoveredDirtyJobResults,
+ recoveredApplications,
+ recoveredDirtyApplicationResults,
executionPlanStore,
- jobResultStore);
+ jobResultStore,
+ applicationStore,
+ applicationResultStore);
completeDispatcherSetup(dispatcherService);
}
- private CompletableFuture<Void>
- createDispatcherBasedOnRecoveredExecutionPlansAndRecoveredDirtyJobResults() {
- // TODO support application recovery which may require fetching user jar from blob server
-
- final CompletableFuture<Collection<JobResult>> dirtyJobsFuture =
+ private CompletableFuture<Void> createDispatcherBasedOnRecoveredApplicationsAndJobs() {
+ final CompletableFuture<Collection<JobResult>> dirtyJobResultsFuture =
CompletableFuture.supplyAsync(this::getDirtyJobResultsIfRunning, ioExecutor);
- return dirtyJobsFuture
- .thenApplyAsync(
- dirtyJobs ->
- this.recoverJobsIfRunning(
- dirtyJobs.stream()
+ final CompletableFuture<Collection<ExecutionPlan>> recoveredJobsFuture =
+ dirtyJobResultsFuture.thenApplyAsync(
+ dirtyJobResults ->
+ recoverJobsIfRunning(
+ dirtyJobResults.stream()
.map(JobResult::getJobId)
.collect(Collectors.toSet())),
- ioExecutor)
- .thenAcceptBoth(dirtyJobsFuture, this::createDispatcherIfRunning)
+ ioExecutor);
+
+ final CompletableFuture<Collection<ApplicationResult>> dirtyApplicationResultsFuture =
+ CompletableFuture.supplyAsync(
+ this::getDirtyApplicationResultsIfRunning, ioExecutor);
+
+ final CompletableFuture<Collection<AbstractApplication>> recoveredApplicationsFuture =
+ dirtyApplicationResultsFuture.thenCombineAsync(
+ recoveredJobsFuture,
+ (dirtyApplicationResults, recoveredJobs) -> {
+ return recoverApplicationsIfRunning(
+ dirtyApplicationResults.stream()
+ .map(ApplicationResult::getApplicationId)
+ .collect(Collectors.toSet()),
+ recoveredJobsFuture.join(),
Review Comment:
Updated: use `thenCompose` instead of `thenCombine` to properly handle futures that complete exceptionally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org