[FLINK-17796] Respect user specified classpath for application mode

This closes apache#12222.
derekm · May 18, 2020 · b591f90 · b591f90
1 parent 2f13ba1
commit b591f90
Show file tree

Hide file tree

Showing 4 changed files with 43 additions and 23 deletions.
diff --git a/...ain/java/org/apache/flink/client/deployment/application/ApplicationClusterEntryPoint.java b/...ain/java/org/apache/flink/client/deployment/application/ApplicationClusterEntryPoint.java
@@ -18,8 +18,12 @@
 
 package org.apache.flink.client.deployment.application;
 
+import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor;
 import org.apache.flink.client.program.PackagedProgram;
+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.runtime.concurrent.ScheduledExecutor;
 import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
 import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
@@ -31,6 +35,12 @@
 import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
 import org.apache.flink.runtime.rest.JobRestEndpointFactory;
 
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
@@ -69,4 +79,19 @@ protected ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
  final ScheduledExecutor scheduledExecutor) {
  return new MemoryArchivedExecutionGraphStore();
  }
+
+ protected static void configureExecution(final Configuration configuration, final PackagedProgram program) throws MalformedURLException {
+ configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, program.getJobJarAndDependencies(), URL::toString);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, getClasspath(configuration, program), URL::toString);
+ }
+
+ private static List<URL> getClasspath(final Configuration configuration, final PackagedProgram program) throws MalformedURLException {
+ final List<URL> classpath = ConfigUtils.decodeListFromConfig(
+ configuration,
+ PipelineOptions.CLASSPATHS,
+ URL::new);
+ classpath.addAll(program.getClasspaths());
+ return Collections.unmodifiableList(classpath.stream().distinct().collect(Collectors.toList()));
+ }
 }
diff --git a/...in/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java b/...in/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
@@ -23,13 +23,9 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint;
 import org.apache.flink.client.deployment.application.ClassPathPackagedProgramRetriever;
-import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor;
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.PackagedProgramRetriever;
-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.configuration.PipelineOptionsInternal;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
@@ -44,7 +40,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.net.URL;
 
 import static org.apache.flink.runtime.util.ClusterEntrypointUtils.tryFindUserLibDirectory;
 
@@ -87,9 +82,12 @@ public static void main(String[] args) {
  }
 
  Configuration configuration = loadConfigurationFromClusterConfig(clusterConfiguration);
- configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, program.getJobJarAndDependencies(), URL::toString);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, program.getClasspaths(), URL::toString);
+ try {
+ configureExecution(configuration, program);
+ } catch (Exception e) {
+ LOG.error("Could not apply application configuration.", e);
+ System.exit(1);
+ }
 
  StandaloneApplicationClusterEntryPoint entrypoint = new StandaloneApplicationClusterEntryPoint(configuration, program);
 

diff --git a/...n/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java b/...n/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java
@@ -22,13 +22,9 @@
 import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint;
 import org.apache.flink.client.deployment.application.ApplicationConfiguration;
 import org.apache.flink.client.deployment.application.ClassPathPackagedProgramRetriever;
-import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor;
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.PackagedProgramRetriever;
-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.kubernetes.utils.KubernetesUtils;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.util.EnvironmentInformation;
@@ -41,7 +37,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.net.URL;
 import java.util.List;
 
 import static org.apache.flink.runtime.util.ClusterEntrypointUtils.tryFindUserLibDirectory;
@@ -74,9 +69,12 @@ public static void main(final String[] args) {
  System.exit(1);
  }
 
- configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, program.getJobJarAndDependencies(), URL::toString);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, program.getClasspaths(), URL::toString);
+ try {
+ configureExecution(configuration, program);
+ } catch (Exception e) {
+ LOG.error("Could not apply application configuration.", e);
+ System.exit(1);
+ }
 
  final KubernetesApplicationClusterEntrypoint kubernetesApplicationClusterEntrypoint =
  new KubernetesApplicationClusterEntrypoint(configuration, program);

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
@@ -22,12 +22,9 @@
 import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint;
 import org.apache.flink.client.deployment.application.ApplicationConfiguration;
 import org.apache.flink.client.deployment.application.ClassPathPackagedProgramRetriever;
-import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor;
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.PackagedProgramRetriever;
-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.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.util.EnvironmentInformation;
@@ -44,7 +41,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.net.URL;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -96,9 +92,12 @@ public static void main(final String[] args) {
  System.exit(1);
  }
 
- configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, program.getJobJarAndDependencies(), URL::toString);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, program.getClasspaths(), URL::toString);
+ try {
+ configureExecution(configuration, program);
+ } catch (Exception e) {
+ LOG.error("Could not apply application configuration.", e);
+ System.exit(1);
+ }
 
  YarnApplicationClusterEntryPoint yarnApplicationClusterEntrypoint =
  new YarnApplicationClusterEntryPoint(configuration, program);