[minor] Refactor getJarFiles() out of ScalaShellEnvironments

deepfield · Dec 7, 2019 · 81bd77d · 81bd77d
1 parent 4d33f4f
commit 81bd77d
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 44 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/util/JarUtils.java b/flink-core/src/main/java/org/apache/flink/util/JarUtils.java
@@ -22,9 +22,14 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.MalformedURLException;
 import java.net.URISyntaxException;
 import java.net.URL;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.jar.JarFile;
+import java.util.stream.Collectors;
 
 /**
  * Utility functions for jar files.
@@ -52,4 +57,22 @@ public static void checkJarFile(URL jar) throws IOException {
  throw new IOException("Error while opening jar file '" + jarFile.getAbsolutePath() + '\'', e);
  }
  }
+
+ public static List<URL> getJarFiles(final String[] jars) {
+ if (jars == null) {
+ return Collections.emptyList();
+ }
+
+ return Arrays.stream(jars).map(jarPath -> {
+ try {
+ final URL fileURL = new File(jarPath).getAbsoluteFile().toURI().toURL();
+ JarUtils.checkJarFile(fileURL);
+ return fileURL;
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException("JAR file path invalid", e);
+ } catch (IOException e) {
+ throw new RuntimeException("Problem with jar file " + jarPath, e);
+ }
+ }).collect(Collectors.toList());
+ }
 }
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
@@ -28,15 +28,10 @@
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.util.JarUtils;
 
-import java.io.File;
-import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 import static org.apache.flink.util.Preconditions.checkState;
@@ -63,7 +58,7 @@ public ScalaShellEnvironment(
 
  super(validateAndGetConfiguration(configuration));
  this.flinkILoop = checkNotNull(flinkILoop);
- this.jarFiles = checkNotNull(getJarFiles(jarFiles));
+ this.jarFiles = checkNotNull(JarUtils.getJarFiles(jarFiles));
  }
 
  private static Configuration validateAndGetConfiguration(final Configuration configuration) {
@@ -75,22 +70,6 @@ private static Configuration validateAndGetConfiguration(final Configuration con
  return checkNotNull(configuration);
  }
 
- private static List<URL> getJarFiles(final String[] jars) {
- return jars == null
- ? Collections.emptyList()
- : Arrays.stream(jars).map(jarPath -> {
- try {
- final URL fileURL = new File(jarPath).getAbsoluteFile().toURI().toURL();
- JarUtils.checkJarFile(fileURL);
- return fileURL;
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException("JAR file path invalid", e);
- } catch (IOException e) {
- throw new RuntimeException("Problem with jar file " + jarPath, e);
- }
- }).collect(Collectors.toList());
- }
-
  @Override
  public JobExecutionResult execute(String jobName) throws Exception {
  final Configuration configuration = getConfiguration();

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
@@ -30,15 +30,10 @@
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.util.JarUtils;
 
-import java.io.File;
-import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 import static org.apache.flink.util.Preconditions.checkState;
@@ -62,7 +57,7 @@ public ScalaShellStreamEnvironment(
 
  super(validateAndGetConfiguration(configuration));
  this.flinkILoop = checkNotNull(flinkILoop);
- this.jarFiles = checkNotNull(getJarFiles(jarFiles));
+ this.jarFiles = checkNotNull(JarUtils.getJarFiles(jarFiles));
  }
 
  private static Configuration validateAndGetConfiguration(final Configuration configuration) {
@@ -74,22 +69,6 @@ private static Configuration validateAndGetConfiguration(final Configuration con
  return checkNotNull(configuration);
  }
 
- private static List<URL> getJarFiles(final String[] jars) {
- return jars == null
- ? Collections.emptyList()
- : Arrays.stream(jars).map(jarPath -> {
- try {
- final URL fileURL = new File(jarPath).getAbsoluteFile().toURI().toURL();
- JarUtils.checkJarFile(fileURL);
- return fileURL;
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException("JAR file path invalid", e);
- } catch (IOException e) {
- throw new RuntimeException("Problem with jar file " + jarPath, e);
- }
- }).collect(Collectors.toList());
- }
-
  @Override
  public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
  final Configuration configuration = getConfiguration();