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/10 08:05:25 UTC
[flink] 02/04: [FLINK-15129] Return JobClient instead of JobClient
Future from executeAsync()
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 343daba7f0a8ddb9f1cb6967c586a9ab164163b4
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Mon Dec 9 16:14:19 2019 +0100
[FLINK-15129] Return JobClient instead of JobClient Future from executeAsync()
---
.../flink/client/program/ContextEnvironment.java | 2 +-
.../apache/flink/api/java/ExecutionEnvironment.java | 14 ++++++++------
.../apache/flink/api/scala/ExecutionEnvironment.scala | 8 ++++----
.../api/environment/StreamContextEnvironment.java | 2 +-
.../api/environment/StreamExecutionEnvironment.java | 18 ++++++++++--------
.../api/scala/StreamExecutionEnvironment.scala | 8 ++++----
6 files changed, 28 insertions(+), 24 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
index 15ce9b2..7e8de5b 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
@@ -70,7 +70,7 @@ public class ContextEnvironment extends ExecutionEnvironment {
public JobExecutionResult execute(String jobName) throws Exception {
verifyExecuteIsCalledOnceWhenInDetachedMode();
- JobClient jobClient = executeAsync(jobName).get();
+ JobClient jobClient = executeAsync(jobName);
JobExecutionResult jobExecutionResult;
if (getConfiguration().getBoolean(DeploymentOptions.ATTACHED)) {
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 e06b103..cf82c22 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
@@ -817,7 +817,7 @@ public class ExecutionEnvironment {
* @throws Exception Thrown, if the program executions fails.
*/
public JobExecutionResult execute(String jobName) throws Exception {
- final JobClient jobClient = executeAsync(jobName).get();
+ final JobClient jobClient = executeAsync(jobName);
if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
lastJobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();
@@ -837,11 +837,11 @@ public class ExecutionEnvironment {
*
* <p>The program execution will be logged and displayed with a generated default name.
*
- * @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
+ * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception Thrown, if the program submission fails.
*/
@PublicEvolving
- public final CompletableFuture<JobClient> executeAsync() throws Exception {
+ public final JobClient executeAsync() throws Exception {
return executeAsync(getDefaultName());
}
@@ -854,11 +854,11 @@ public class ExecutionEnvironment {
*
* <p>The program execution will be logged and displayed with the given job name.
*
- * @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
+ * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception Thrown, if the program submission fails.
*/
@PublicEvolving
- public CompletableFuture<JobClient> executeAsync(String jobName) throws Exception {
+ public JobClient executeAsync(String jobName) throws Exception {
checkNotNull(configuration.get(DeploymentOptions.TARGET), "No execution.target specified in your configuration file.");
consolidateParallelismDefinitionsInConfiguration();
@@ -872,9 +872,11 @@ public class ExecutionEnvironment {
"Cannot find compatible factory for specified execution.target (=%s)",
configuration.get(DeploymentOptions.TARGET));
- return executorFactory
+ CompletableFuture<JobClient> jobClientFuture = executorFactory
.getExecutor(configuration)
.execute(plan, configuration);
+
+ return jobClientFuture.get();
}
private void consolidateParallelismDefinitionsInConfiguration() {
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index 479f876..9c5598d 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -508,11 +508,11 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
* its usage. In other case, there may be resource leaks depending on the JobClient
* implementation.
*
- * @return A future of [[JobClient]] that can be used to communicate with the submitted job,
+ * @return A [[JobClient]] that can be used to communicate with the submitted job,
* completed on submission succeeded.
*/
@PublicEvolving
- def executeAsync(): CompletableFuture[JobClient] = javaEnv.executeAsync()
+ def executeAsync(): JobClient = javaEnv.executeAsync()
/**
* Triggers the program execution asynchronously.
@@ -528,11 +528,11 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
* its usage. In other case, there may be resource leaks depending on the JobClient
* implementation.
*
- * @return A future of [[JobClient]] that can be used to communicate with the submitted job,
+ * @return A [[JobClient]] that can be used to communicate with the submitted job,
* completed on submission succeeded.
*/
@PublicEvolving
- def executeAsync(jobName: String): CompletableFuture[JobClient] = javaEnv.executeAsync(jobName)
+ def executeAsync(jobName: String): JobClient = javaEnv.executeAsync(jobName)
/**
* Creates the plan with which the system will execute the program, and returns it as a String
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
index 23d7c78..ba772e0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
@@ -62,7 +62,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
transformations.clear();
- JobClient jobClient = executeAsync(streamGraph).get();
+ JobClient jobClient = executeAsync(streamGraph);
JobExecutionResult jobExecutionResult;
if (getConfiguration().getBoolean(DeploymentOptions.ATTACHED)) {
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 3cd448e..34aab2e 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
@@ -1635,7 +1635,7 @@ public class StreamExecutionEnvironment {
*/
@Internal
public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
- final JobClient jobClient = executeAsync(streamGraph).get();
+ final JobClient jobClient = executeAsync(streamGraph);
if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
return jobClient.getJobExecutionResult(userClassloader).get();
@@ -1652,11 +1652,11 @@ public class StreamExecutionEnvironment {
* <p>The program execution will be logged and displayed with a generated
* default name.
*
- * @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
+ * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception which occurs during job execution.
*/
@PublicEvolving
- public final CompletableFuture<JobClient> executeAsync() throws Exception {
+ public final JobClient executeAsync() throws Exception {
return executeAsync(DEFAULT_JOB_NAME);
}
@@ -1668,11 +1668,11 @@ public class StreamExecutionEnvironment {
* <p>The program execution will be logged and displayed with the provided name
*
* @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.
+ * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception which occurs during job execution.
*/
@PublicEvolving
- public CompletableFuture<JobClient> executeAsync(String jobName) throws Exception {
+ public JobClient executeAsync(String jobName) throws Exception {
return executeAsync(getStreamGraph(checkNotNull(jobName)));
}
@@ -1682,11 +1682,11 @@ public class StreamExecutionEnvironment {
* for example printing results or forwarding them to a message queue.
*
* @param streamGraph the stream graph representing the transformations
- * @return A future of {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
+ * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
* @throws Exception which occurs during job execution.
*/
@Internal
- public CompletableFuture<JobClient> executeAsync(StreamGraph streamGraph) throws Exception {
+ public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
checkNotNull(streamGraph, "StreamGraph cannot be null.");
checkNotNull(configuration.get(DeploymentOptions.TARGET), "No execution.target specified in your configuration file.");
@@ -1700,9 +1700,11 @@ public class StreamExecutionEnvironment {
"Cannot find compatible factory for specified execution.target (=%s)",
configuration.get(DeploymentOptions.TARGET));
- return executorFactory
+ CompletableFuture<JobClient> jobClientFuture = executorFactory
.getExecutor(configuration)
.execute(streamGraph, configuration);
+
+ return jobClientFuture.get();
}
private void consolidateParallelismDefinitionsInConfiguration() {
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index 2ce8ce7..f911f3e 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -692,11 +692,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
* its usage. In other case, there may be resource leaks depending on the JobClient
* implementation.
*
- * @return A future of [[JobClient]] that can be used to communicate with the submitted job,
+ * @return A [[JobClient]] that can be used to communicate with the submitted job,
* completed on submission succeeded.
*/
@PublicEvolving
- def executeAsync(): CompletableFuture[JobClient] = javaEnv.executeAsync()
+ def executeAsync(): JobClient = javaEnv.executeAsync()
/**
* Triggers the program execution asynchronously. The environment will execute all parts of
@@ -710,11 +710,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
* its usage. In other case, there may be resource leaks depending on the JobClient
* implementation.
*
- * @return A future of [[JobClient]] that can be used to communicate with the submitted job,
+ * @return A [[JobClient]] that can be used to communicate with the submitted job,
* completed on submission succeeded.
*/
@PublicEvolving
- def executeAsync(jobName: String): CompletableFuture[JobClient] = javaEnv.executeAsync(jobName)
+ def executeAsync(jobName: String): JobClient = javaEnv.executeAsync(jobName)
/**
* Creates the plan with which the system will execute the program, and