[FLINK-17870] Fix ScalaShell executeAsync to ship all dependencies

This closes apache#12288.
deepfield · May 25, 2020 · ea54a54 · ea54a54
1 parent 42077e9
commit ea54a54
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 8 deletions.
diff --git a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellEnvironment.java b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellEnvironment.java
@@ -20,12 +20,12 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.InvalidProgramException;
-import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.scala.FlinkILoop;
 import org.apache.flink.configuration.ConfigUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.util.JarUtils;
 
 import java.net.MalformedURLException;
@@ -71,14 +71,17 @@ private static Configuration validateAndGetConfiguration(final Configuration con
  }
 
  @Override
- public JobExecutionResult execute(String jobName) throws Exception {
+ public JobClient executeAsync(String jobName) throws Exception {
+ updateDependencies();
+ return super.executeAsync(jobName);
+ }
+
+ private void updateDependencies() throws Exception {
  final Configuration configuration = getConfiguration();
  checkState(configuration.getBoolean(DeploymentOptions.ATTACHED), "Only ATTACHED mode is supported by the scala shell.");
 
  final List<URL> updatedJarFiles = getUpdatedJarFiles();
  ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, updatedJarFiles, URL::toString);
-
- return super.execute(jobName);
  }
 
  private List<URL> getUpdatedJarFiles() throws MalformedURLException {

diff --git a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellStreamEnvironment.java b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellStreamEnvironment.java
@@ -20,12 +20,12 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.InvalidProgramException;
-import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.scala.FlinkILoop;
 import org.apache.flink.configuration.ConfigUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.util.JarUtils;
@@ -70,14 +70,17 @@ private static Configuration validateAndGetConfiguration(final Configuration con
  }
 
  @Override
- public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
+ public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
+ updateDependencies();
+ return super.executeAsync(streamGraph);
+ }
+
+ private void updateDependencies() throws Exception {
  final Configuration configuration = getConfiguration();
  checkState(configuration.getBoolean(DeploymentOptions.ATTACHED), "Only ATTACHED mode is supported by the scala shell.");
 
  final List<URL> updatedJarFiles = getUpdatedJarFiles();
  ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, updatedJarFiles, URL::toString);
-
- return super.execute(streamGraph);
  }
 
  public Configuration getClientConfiguration() {