[minor] Refactoring the ClassPathPackagedProgramRetriever creation

mbalassi · May 15, 2020 · 8c52ea0 · 8c52ea0
1 parent 7bea158
commit 8c52ea0
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 9 deletions.
diff --git a/...in/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java b/...in/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
@@ -42,6 +42,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.File;
 import java.io.IOException;
 import java.net.URL;
 
@@ -114,11 +115,12 @@ private static PackagedProgram getPackagedProgram(
  private static PackagedProgramRetriever getPackagedProgramRetriever(
  final String[] programArguments,
  @Nullable final String jobClassName) throws IOException {
+ final File userLibDir = tryFindUserLibDirectory().orElse(null);
  final ClassPathPackagedProgramRetriever.Builder retrieverBuilder =
  ClassPathPackagedProgramRetriever
  .newBuilder(programArguments)
+ .setUserLibDirectory(userLibDir)
  .setJobClassName(jobClassName);
- tryFindUserLibDirectory().ifPresent(retrieverBuilder::setUserLibDirectory);
  return retrieverBuilder.build();
  }
 

diff --git a/...n/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java b/...n/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java
@@ -104,12 +104,14 @@ private static PackagedProgramRetriever getPackagedProgramRetriever(
  final List<File> pipelineJars = KubernetesUtils.checkJarFileForApplicationMode(configuration);
  Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
 
+ final File userLibDir = tryFindUserLibDirectory().orElse(null);
+
  final ClassPathPackagedProgramRetriever.Builder retrieverBuilder =
  ClassPathPackagedProgramRetriever
  .newBuilder(programArguments)
+ .setUserLibDirectory(userLibDir)
  .setJarFile(pipelineJars.get(0))
  .setJobClassName(jobClassName);
- tryFindUserLibDirectory().ifPresent(retrieverBuilder::setUserLibDirectory);
  return retrieverBuilder.build();
  }
 }
diff --git a/...yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java b/...yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
@@ -123,17 +123,23 @@ private static PackagedProgramRetriever getPackagedProgramRetriever(
  final String[] programArguments,
  @Nullable final String jobClassName) throws IOException {
 
- final List<File> pipelineJars = configuration.get(PipelineOptions.JARS).stream()
- .map(uri -> new File(YarnEntrypointUtils.getUsrLibDir(configuration).orElse(null), new Path(uri).getName()))
- .collect(Collectors.toList());
- Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
-
+ final File userLibDir = YarnEntrypointUtils.getUsrLibDir(configuration).orElse(null);
+ final File userApplicationJar = getUserApplicationJar(userLibDir, configuration);
  final ClassPathPackagedProgramRetriever.Builder retrieverBuilder =
  ClassPathPackagedProgramRetriever
  .newBuilder(programArguments)
- .setJarFile(pipelineJars.get(0))
+ .setUserLibDirectory(userLibDir)
+ .setJarFile(userApplicationJar)
  .setJobClassName(jobClassName);
- YarnEntrypointUtils.getUsrLibDir(configuration).ifPresent(retrieverBuilder::setUserLibDirectory);
  return retrieverBuilder.build();
  }
+
+ private static File getUserApplicationJar(final File userLibDir, final Configuration configuration) {
+ final List<File> pipelineJars = configuration.get(PipelineOptions.JARS).stream()
+ .map(uri -> new File(userLibDir, new Path(uri).getName()))
+ .collect(Collectors.toList());
+
+ Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
+ return pipelineJars.get(0);
+ }
 }