You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/12/08 15:13:05 UTC
[flink] 05/06: [FLINK-15116] Make JobClient stateless,
remove AutoCloseable
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1b99d5383d831025b19cea965abc888199c5f2b7
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Fri Dec 6 16:24:53 2019 +0100
[FLINK-15116] Make JobClient stateless, remove AutoCloseable
With this change, the JobClient acquires the required ClusterClient for
each method call. This means that the users no longer have the burden of
managing the JobClient lifecycle, i.e. they can freely ignore the result
of executeAsync().
---
.../deployment/AbstractJobClusterExecutor.java | 34 ++++----
.../deployment/AbstractSessionClusterExecutor.java | 17 ++--
.../deployment/ClusterClientJobClientAdapter.java | 97 ++++++++++++----------
.../client/deployment/executors/LocalExecutor.java | 18 ++--
.../apache/flink/client/program/ClientTest.java | 2 +-
.../org/apache/flink/core/execution/JobClient.java | 5 +-
.../flink/api/java/ExecutionEnvironment.java | 20 ++---
.../apache/flink/api/java/TestingJobClient.java | 5 --
.../environment/StreamExecutionEnvironment.java | 18 ++--
.../RemoteStreamExecutionEnvironmentTest.java | 2 +-
.../streaming/environment/TestingJobClient.java | 5 --
.../table/client/gateway/local/LocalExecutor.java | 3 +-
.../gateway/local/result/CollectStreamResult.java | 26 +-----
.../result/MaterializedCollectBatchResult.java | 29 +------
14 files changed, 107 insertions(+), 174 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractJobClusterExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractJobClusterExecutor.java
index fb603bc..f564f07 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractJobClusterExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractJobClusterExecutor.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.cli.ExecutionConfigAccessor;
import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.Executor;
import org.apache.flink.core.execution.JobClient;
@@ -63,31 +64,28 @@ public class AbstractJobClusterExecutor<ClusterID, ClientFactory extends Cluster
final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);
- final ClusterClient<ClusterID> clusterClient = clusterDescriptor
- .deployJobCluster(clusterSpecification, jobGraph, configAccessor.getDetachedMode())
- .getClusterClient();
+ final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor
+ .deployJobCluster(clusterSpecification, jobGraph, configAccessor.getDetachedMode());
LOG.info("Job has been submitted with JobID " + jobGraph.getJobID());
final boolean withShutdownHook = !configAccessor.getDetachedMode() && configAccessor.isShutdownOnAttachedExit();
if (withShutdownHook) {
- Thread shutdownHook = ShutdownHookUtil.addShutdownHook(
- clusterClient::shutDownCluster, clusterClient.getClass().getSimpleName(), LOG);
-
- return CompletableFuture.completedFuture(new ClusterClientJobClientAdapter<ClusterID>(clusterClient, jobGraph.getJobID()) {
- @Override
- protected void doClose() {
- ShutdownHookUtil.removeShutdownHook(shutdownHook, clusterClient.getClass().getSimpleName(), LOG);
- clusterClient.close();
- }
+ ShutdownHookUtil.addShutdownHook(
+ () -> {
+ try (ClusterClient<ClusterID> client = clusterClientProvider.getClusterClient()) {
+ client.shutDownCluster();
+ }
+ },
+ "Cluster shutdown hook for attached Job execution",
+ LOG);
+
+ return CompletableFuture.completedFuture(
+ new ClusterClientJobClientAdapter<ClusterID>(clusterClientProvider, jobGraph.getJobID()) {
});
} else {
- return CompletableFuture.completedFuture(new ClusterClientJobClientAdapter<ClusterID>(clusterClient, jobGraph.getJobID()) {
- @Override
- protected void doClose() {
- clusterClient.close();
- }
- });
+ return CompletableFuture.completedFuture(
+ new ClusterClientJobClientAdapter<>(clusterClientProvider, jobGraph.getJobID()));
}
}
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractSessionClusterExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractSessionClusterExecutor.java
index 784edcd..3682fdc 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractSessionClusterExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractSessionClusterExecutor.java
@@ -21,6 +21,7 @@ package org.apache.flink.client.deployment;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.Executor;
import org.apache.flink.core.execution.JobClient;
@@ -56,18 +57,14 @@ public class AbstractSessionClusterExecutor<ClusterID, ClientFactory extends Clu
final ClusterID clusterID = clusterClientFactory.getClusterId(configuration);
checkState(clusterID != null);
- final ClusterClient<ClusterID> clusterClient = clusterDescriptor
- .retrieve(clusterID)
- .getClusterClient();
-
+ final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor.retrieve(clusterID);
+ ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();
return clusterClient
.submitJob(jobGraph)
- .thenApply(jobID -> new ClusterClientJobClientAdapter<ClusterID>(clusterClient, jobID) {
- @Override
- protected void doClose() {
- clusterClient.close();
- }
- });
+ .thenApplyAsync(jobID -> (JobClient) new ClusterClientJobClientAdapter<>(
+ clusterClientProvider,
+ jobID))
+ .whenComplete((ignored1, ignored2) -> clusterClient.close());
}
}
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java
index 73a343f..3fb0e5c 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java
@@ -22,20 +22,19 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.jobmaster.JobResult;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.function.FunctionUtils;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+
+import org.apache.commons.io.IOUtils;
import javax.annotation.Nullable;
-import java.io.IOException;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -44,15 +43,13 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public class ClusterClientJobClientAdapter<ClusterID> implements JobClient {
- private final ClusterClient<ClusterID> clusterClient;
+ private final ClusterClientProvider<ClusterID> clusterClientProvider;
private final JobID jobID;
- private final AtomicBoolean running = new AtomicBoolean(true);
-
- public ClusterClientJobClientAdapter(final ClusterClient<ClusterID> clusterClient, final JobID jobID) {
+ public ClusterClientJobClientAdapter(final ClusterClientProvider<ClusterID> clusterClientProvider, final JobID jobID) {
this.jobID = checkNotNull(jobID);
- this.clusterClient = checkNotNull(clusterClient);
+ this.clusterClientProvider = checkNotNull(clusterClientProvider);
}
@Override
@@ -62,62 +59,78 @@ public class ClusterClientJobClientAdapter<ClusterID> implements JobClient {
@Override
public CompletableFuture<JobStatus> getJobStatus() {
- return clusterClient.getJobStatus(jobID);
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient -> clusterClient.getJobStatus(jobID)));
}
@Override
public CompletableFuture<Void> cancel() {
- return clusterClient.cancel(jobID).thenApply(FunctionUtils.nullFn());
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient -> clusterClient.cancel(jobID).thenApply((ignored) -> null)));
}
@Override
public CompletableFuture<String> stopWithSavepoint(boolean advanceToEndOfEventTime, @Nullable String savepointDirectory) {
- return clusterClient.stopWithSavepoint(jobID, advanceToEndOfEventTime, savepointDirectory);
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient ->
+ clusterClient.stopWithSavepoint(jobID, advanceToEndOfEventTime, savepointDirectory)));
}
@Override
public CompletableFuture<String> triggerSavepoint(@Nullable String savepointDirectory) {
- return clusterClient.triggerSavepoint(jobID, savepointDirectory);
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient ->
+ clusterClient.triggerSavepoint(jobID, savepointDirectory)));
}
@Override
public CompletableFuture<Map<String, Object>> getAccumulators(ClassLoader classLoader) {
- return clusterClient.getAccumulators(jobID, classLoader);
+ checkNotNull(classLoader);
+
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient ->
+ clusterClient.getAccumulators(jobID, classLoader)));
}
@Override
public CompletableFuture<JobExecutionResult> getJobExecutionResult(final ClassLoader userClassloader) {
checkNotNull(userClassloader);
- final CompletableFuture<JobResult> jobResultFuture = clusterClient.requestJobResult(jobID);
- return jobResultFuture.handle((jobResult, throwable) -> {
- if (throwable != null) {
- ExceptionUtils.checkInterrupted(throwable);
- throw new CompletionException(new ProgramInvocationException("Could not run job", jobID, throwable));
- } else {
- try {
- return jobResult.toJobExecutionResult(userClassloader);
- } catch (JobExecutionException | IOException | ClassNotFoundException e) {
- throw new CompletionException(new ProgramInvocationException("Job failed", jobID, e));
- }
- }
- });
+ return bridgeClientRequest(
+ clusterClientProvider,
+ (clusterClient -> clusterClient
+ .requestJobResult(jobID)
+ .thenApply((jobResult) -> {
+ try {
+ return jobResult.toJobExecutionResult(userClassloader);
+ } catch (Throwable t) {
+ throw new CompletionException(
+ new ProgramInvocationException("Job failed", jobID, t));
+ }
+ })));
}
- @Override
- public final void close() {
- if (running.compareAndSet(true, false)) {
- doClose();
- }
- }
+ private static <T> CompletableFuture<T> bridgeClientRequest(
+ ClusterClientProvider<?> clusterClientProvider,
+ Function<ClusterClient<?>, CompletableFuture<T>> resultRetriever) {
+
+ ClusterClient<?> clusterClient = clusterClientProvider.getClusterClient();
- /**
- * Method to be overridden by subclass which contains actual close actions.
- *
- * <p>We do close in this way to ensure multiple calls to {@link #close()}
- * are executed at most once guarded by {@link #running} flag.
- */
- protected void doClose() {
+ CompletableFuture<T> resultFuture;
+ try {
+ resultFuture = resultRetriever.apply(clusterClient);
+ } catch (Throwable throwable) {
+ IOUtils.closeQuietly(clusterClient::close);
+ return FutureUtils.completedExceptionally(throwable);
+ }
+ return resultFuture.whenCompleteAsync(
+ (jobResult, throwable) -> IOUtils.closeQuietly(clusterClient::close));
}
+
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
index ec3632c..1f4e6b2 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/LocalExecutor.java
@@ -19,6 +19,7 @@
package org.apache.flink.client.deployment.executors;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
@@ -62,15 +63,14 @@ public class LocalExecutor implements Executor {
final MiniCluster miniCluster = startMiniCluster(jobGraph, configuration);
final MiniClusterClient clusterClient = new MiniClusterClient(configuration, miniCluster);
- return clusterClient
- .submitJob(jobGraph)
- .thenApply(jobID -> new ClusterClientJobClientAdapter<MiniClusterClient.MiniClusterId>(clusterClient, jobID) {
- @Override
- protected void doClose() {
- clusterClient.close();
- shutdownMiniCluster(miniCluster);
- }
- });
+ CompletableFuture<JobID> jobIdFuture = clusterClient.submitJob(jobGraph);
+
+ jobIdFuture
+ .thenCompose(clusterClient::requestJobResult)
+ .thenAccept((jobResult) -> clusterClient.shutDownCluster());
+
+ return jobIdFuture.thenApply(jobID ->
+ new ClusterClientJobClientAdapter<>(() -> clusterClient, jobID));
}
private JobGraph getJobGraph(Pipeline pipeline, Configuration configuration) {
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index 5aa9f21..dc140f8 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -396,7 +396,7 @@ public class ClientTest extends TestLogger {
jobGraph.setClasspaths(accessor.getClasspaths());
final JobID jobID = ClientUtils.submitJob(clusterClient, jobGraph).getJobID();
- return CompletableFuture.completedFuture(new ClusterClientJobClientAdapter<>(clusterClient, jobID));
+ return CompletableFuture.completedFuture(new ClusterClientJobClientAdapter<>(() -> clusterClient, jobID));
};
}
};
diff --git a/flink-core/src/main/java/org/apache/flink/core/execution/JobClient.java b/flink-core/src/main/java/org/apache/flink/core/execution/JobClient.java
index d5ee6bc..084d61a 100644
--- a/flink-core/src/main/java/org/apache/flink/core/execution/JobClient.java
+++ b/flink-core/src/main/java/org/apache/flink/core/execution/JobClient.java
@@ -32,7 +32,7 @@ import java.util.concurrent.CompletableFuture;
* A client that is scoped to a specific job.
*/
@PublicEvolving
-public interface JobClient extends AutoCloseable {
+public interface JobClient {
/**
* Returns the {@link JobID} that uniquely identifies the job this client is scoped to.
@@ -83,7 +83,4 @@ public interface JobClient extends AutoCloseable {
* @param userClassloader the classloader used to de-serialize the accumulators of the job.
*/
CompletableFuture<JobExecutionResult> getJobExecutionResult(final ClassLoader userClassloader);
-
- @Override
- void close();
}
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index 24072ee..a8705a4 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -804,13 +804,15 @@ public class ExecutionEnvironment {
* @throws Exception Thrown, if the program executions fails.
*/
public JobExecutionResult execute(String jobName) throws Exception {
- try (final JobClient jobClient = executeAsync(jobName).get()) {
- lastJobExecutionResult = configuration.getBoolean(DeploymentOptions.ATTACHED)
- ? jobClient.getJobExecutionResult(userClassloader).get()
- : new DetachedJobExecutionResult(jobClient.getJobID());
+ final JobClient jobClient = executeAsync(jobName).get();
- return lastJobExecutionResult;
+ if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
+ lastJobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();
+ } else {
+ lastJobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
}
+
+ return lastJobExecutionResult;
}
/**
@@ -822,10 +824,6 @@ public class ExecutionEnvironment {
*
* <p>The program execution will be logged and displayed with a generated default name.
*
- * <p><b>ATTENTION:</b> The caller of this method is responsible for managing the lifecycle of
- * the returned {@link JobClient}. This means calling {@link JobClient#close()} at the end of
- * its usage. In other case, there may be resource leaks depending on the JobClient implementation.
- *
* @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception Thrown, if the program submission fails.
*/
@@ -843,10 +841,6 @@ public class ExecutionEnvironment {
*
* <p>The program execution will be logged and displayed with the given job name.
*
- * <p><b>ATTENTION:</b> The caller of this method is responsible for managing the lifecycle of
- * the returned {@link JobClient}. This means calling {@link JobClient#close()} at the end of
- * its usage. In other case, there may be resource leaks depending on the JobClient implementation.
- *
* @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception Thrown, if the program submission fails.
*/
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/TestingJobClient.java b/flink-java/src/test/java/org/apache/flink/api/java/TestingJobClient.java
index 6b7fb11..fefe784 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/TestingJobClient.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/TestingJobClient.java
@@ -68,9 +68,4 @@ public class TestingJobClient implements JobClient {
public CompletableFuture<Map<String, Object>> getAccumulators(ClassLoader classLoader) {
return CompletableFuture.completedFuture(Collections.emptyMap());
}
-
- @Override
- public void close() {
-
- }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 5d02355..9063f83 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -1619,10 +1619,12 @@ public class StreamExecutionEnvironment {
*/
@Internal
public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
- try (final JobClient jobClient = executeAsync(streamGraph).get()) {
- return configuration.getBoolean(DeploymentOptions.ATTACHED)
- ? jobClient.getJobExecutionResult(userClassloader).get()
- : new DetachedJobExecutionResult(jobClient.getJobID());
+ final JobClient jobClient = executeAsync(streamGraph).get();
+
+ if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
+ return jobClient.getJobExecutionResult(userClassloader).get();
+ } else {
+ return new DetachedJobExecutionResult(jobClient.getJobID());
}
}
@@ -1634,10 +1636,6 @@ public class StreamExecutionEnvironment {
* <p>The program execution will be logged and displayed with a generated
* default name.
*
- * <p><b>ATTENTION:</b> The caller of this method is responsible for managing the lifecycle of
- * the returned {@link JobClient}. This means calling {@link JobClient#close()} at the end of
- * its usage. In other case, there may be resource leaks depending on the JobClient implementation.
- *
* @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception which occurs during job execution.
*/
@@ -1653,10 +1651,6 @@ public class StreamExecutionEnvironment {
*
* <p>The program execution will be logged and displayed with the provided name
*
- * <p><b>ATTENTION:</b> The caller of this method is responsible for managing the lifecycle of
- * the returned {@link JobClient}. This means calling {@link JobClient#close()} at the end of
- * its usage. In other case, there may be resource leaks depending on the JobClient implementation.
- *
* @param jobName desired name of the job
* @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception which occurs during job execution.
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/RemoteStreamExecutionEnvironmentTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/RemoteStreamExecutionEnvironmentTest.java
index 309da92..fdf9895 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/RemoteStreamExecutionEnvironmentTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/RemoteStreamExecutionEnvironmentTest.java
@@ -149,7 +149,7 @@ public class RemoteStreamExecutionEnvironmentTest extends TestLogger {
clusterClient = new TestClusterClient(configuration, jobID);
return CompletableFuture.completedFuture(
- new ClusterClientJobClientAdapter<>(clusterClient, jobID));
+ new ClusterClientJobClientAdapter<>(() -> clusterClient, jobID));
};
}
};
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/TestingJobClient.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/TestingJobClient.java
index 30d2435..9e689f5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/TestingJobClient.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/TestingJobClient.java
@@ -68,9 +68,4 @@ public class TestingJobClient implements JobClient {
public CompletableFuture<Map<String, Object>> getAccumulators(ClassLoader classLoader) {
return CompletableFuture.completedFuture(Collections.emptyMap());
}
-
- @Override
- public void close() {
-
- }
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index 8e2b5de..c9f50d0 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -601,7 +601,8 @@ public class LocalExecutor implements Executor {
final ProgramDeployer deployer = new ProgramDeployer(configuration, jobName, pipeline);
// blocking deployment
- try (JobClient jobClient = deployer.deploy().get()) {
+ try {
+ JobClient jobClient = deployer.deploy().get();
return ProgramTargetDescriptor.of(jobClient.getJobID());
} catch (Exception e) {
throw new RuntimeException("Error running SQL job.", e);
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java
index de35dcd..fb7c4b1 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java
@@ -40,7 +40,6 @@ import org.apache.flink.types.Row;
import java.io.IOException;
import java.net.InetAddress;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -101,32 +100,9 @@ public abstract class CollectStreamResult<C> extends BasicResult<C> implements D
// start listener thread
retrievalThread.start();
- // we have to check for exceptions in two places here because job submission can fail
- // at different stages. For batch file sources, the job manager will check if the file
- // exists and job submission will already fail at that early stage, meaning we will not
- // get a job client. For streaming file sources, this check only happens when the job
- // was already successfully submitted.
jobExecutionResultFuture = deployer
.deploy()
- .thenCompose(jobClient -> {
- // fetch the job execution result, so that an attached cluster will shut down
- CompletableFuture<JobExecutionResult> result = jobClient.getJobExecutionResult(
- classLoader);
-
- return result.whenComplete((unusedJobResult, jobException) -> {
- if (jobException != null) {
- executionException.compareAndSet(
- null,
- new SqlExecutionException(
- "Error while submitting job.", jobException));
- }
- try {
- jobClient.close();
- } catch (Exception e) {
- throw new CompletionException("Error while closing JobClient.", e);
- }
- });
- })
+ .thenCompose(jobClient -> jobClient.getJobExecutionResult(classLoader))
.whenComplete((unused, throwable) -> {
if (throwable != null) {
executionException.compareAndSet(
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
index a6c46c1..e944acd 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
@@ -35,8 +35,6 @@ import org.apache.flink.util.AbstractID;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -87,34 +85,9 @@ public class MaterializedCollectBatchResult<C> extends BasicResult<C> implements
@Override
public void startRetrieval(ProgramDeployer deployer) {
- // we have to check for exceptions in two places here because job submission can fail
- // at different stages. For batch file sources, the job manager will check if the file
- // exists and job submission will already fail at that early stage, meaning we will not
- // get a job client. For streaming file sources, this check only happens when the job
- // was already successfully submitted.
deployer
.deploy()
- .thenCompose(jobClient -> {
-
- CompletableFuture<JobExecutionResult> result = jobClient.getJobExecutionResult(
- classLoader);
-
- return result.handle((unusedJobResult, jobException) -> {
- if (jobException != null) {
- executionException.compareAndSet(
- null,
- new SqlExecutionException(
- "Error while submitting job.",
- jobException));
- }
- try {
- jobClient.close();
- } catch (Exception e) {
- throw new CompletionException("Error while closing JobClient.", e);
- }
- return unusedJobResult;
- });
- })
+ .thenCompose(jobClient -> jobClient.getJobExecutionResult(classLoader))
.thenAccept(new ResultRetrievalHandler())
.whenComplete((unused, throwable) -> {
if (throwable != null) {