You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2019/11/14 09:48:13 UTC
[flink] branch master updated (381892f -> 03c125c)
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 381892f [FLINK-14494][docs] Regenerated config options documentation with type information
new f6ac09f [hotfix] Change the signature of the ConfigUtils.encodeCollectionToConfig()
new 86b00ad [hotfix] Move addJars() from ClientUtils to JobGraph
new c2dbf94 [hotfix] Refactors PackagedProgram.extractContainedLibraries()
new cffa2e8 [hotfix] CliFrontend.buildProgram() uses only ProgramOptions
new bd8570f [FLINK-14745] Add dependencies of job as list of URLs in config
new 8293c28 [FLINK-14745] Fix setting CoreOptions.DEFAULT_PARALLELISM in ExecutionConfigAccessor
new 03c125c [FLINK-14745] Wire the configuration to the ClientUtils.executeProgram
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/flink/client/ClientUtils.java | 46 +++----
.../org/apache/flink/client/RemoteExecutor.java | 2 +-
.../org/apache/flink/client/cli/CliFrontend.java | 34 +++--
.../flink/client/cli/ExecutionConfigAccessor.java | 41 ++----
.../flink/client/program/ContextEnvironment.java | 2 +-
.../flink/client/program/PackagedProgram.java | 144 +++++++++------------
.../flink/client/program/PackagedProgramUtils.java | 3 +-
.../client/cli/CliFrontendPackageProgramTest.java | 50 +++----
.../flink/client/cli/CliFrontendRunTest.java | 11 +-
.../apache/flink/client/program/ClientTest.java | 34 ++++-
.../apache/flink/configuration/ConfigUtils.java | 38 ++++--
.../flink/configuration/ConfigUtilsTest.java | 121 +++++++++++++++++
.../apache/flink/runtime/jobgraph/JobGraph.java | 17 +++
.../api/environment/StreamContextEnvironment.java | 2 +-
.../client/gateway/local/ExecutionContext.java | 11 +-
15 files changed, 335 insertions(+), 221 deletions(-)
create mode 100644 flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java
[flink] 04/07: [hotfix] CliFrontend.buildProgram() uses only
ProgramOptions
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cffa2e8647f74e0332cad011d937015ac9ac031e
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Tue Nov 12 09:12:37 2019 +0100
[hotfix] CliFrontend.buildProgram() uses only ProgramOptions
---
.../org/apache/flink/client/cli/CliFrontend.java | 21 +++++----
.../flink/client/cli/ExecutionConfigAccessor.java | 26 -----------
.../client/cli/CliFrontendPackageProgramTest.java | 50 +++++++++-------------
3 files changed, 31 insertions(+), 66 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index 9552978..6e9b2f9 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -185,7 +185,6 @@ public class CliFrontend {
final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, args, true);
final ProgramOptions programOptions = new ProgramOptions(commandLine);
- final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromProgramOptions(programOptions);
// evaluate help flag
if (commandLine.hasOption(HELP_OPTION.getOpt())) {
@@ -195,7 +194,7 @@ public class CliFrontend {
if (!programOptions.isPython()) {
// Java program should be specified a JAR file
- if (executionParameters.getJarFilePath() == null) {
+ if (programOptions.getJarFilePath() == null) {
throw new CliArgsException("Java program should be specified a JAR file.");
}
}
@@ -203,7 +202,7 @@ public class CliFrontend {
final PackagedProgram program;
try {
LOG.info("Building program from JAR file");
- program = buildProgram(programOptions, executionParameters);
+ program = buildProgram(programOptions);
}
catch (FileNotFoundException e) {
throw new CliArgsException("Could not build the program from JAR file.", e);
@@ -211,7 +210,10 @@ public class CliFrontend {
final CustomCommandLine customCommandLine = getActiveCustomCommandLine(commandLine);
final Configuration executorConfig = customCommandLine.applyCommandLineOptionsToConfiguration(commandLine);
+
+ final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromProgramOptions(programOptions);
final Configuration executionConfig = executionParameters.getConfiguration();
+
try {
runProgram(executorConfig, executionConfig, program);
} finally {
@@ -322,7 +324,6 @@ public class CliFrontend {
final CommandLine commandLine = CliFrontendParser.parse(commandOptions, args, true);
final ProgramOptions programOptions = new ProgramOptions(commandLine);
- final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromProgramOptions(programOptions);
// evaluate help flag
if (commandLine.hasOption(HELP_OPTION.getOpt())) {
@@ -337,7 +338,7 @@ public class CliFrontend {
// -------- build the packaged program -------------
LOG.info("Building program from JAR file");
- final PackagedProgram program = buildProgram(programOptions, executionParameters);
+ final PackagedProgram program = buildProgram(programOptions);
try {
int parallelism = programOptions.getParallelism();
@@ -768,12 +769,10 @@ public class CliFrontend {
*
* @return A PackagedProgram (upon success)
*/
- PackagedProgram buildProgram(
- final ProgramOptions runOptions,
- final ExecutionConfigAccessor executionParameters) throws FileNotFoundException, ProgramInvocationException {
+ PackagedProgram buildProgram(final ProgramOptions runOptions) throws FileNotFoundException, ProgramInvocationException {
String[] programArgs = runOptions.getProgramArgs();
- String jarFilePath = executionParameters.getJarFilePath();
- List<URL> classpaths = executionParameters.getClasspaths();
+ String jarFilePath = runOptions.getJarFilePath();
+ List<URL> classpaths = runOptions.getClasspaths();
// Get assembler class
String entryPointClass = runOptions.getEntryPointClassName();
@@ -803,7 +802,7 @@ public class CliFrontend {
.setArguments(programArgs)
.build();
- program.setSavepointRestoreSettings(executionParameters.getSavepointRestoreSettings());
+ program.setSavepointRestoreSettings(runOptions.getSavepointRestoreSettings());
return program;
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
index 77c0440..ee32449 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
@@ -27,10 +27,8 @@ import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-import java.io.File;
import java.net.MalformedURLException;
import java.net.URL;
-import java.util.Collections;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -67,39 +65,15 @@ public class ExecutionConfigAccessor {
ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, options.getClasspaths(), URL::toString);
- parseJarURLToConfig(options.getJarFilePath(), configuration);
-
SavepointRestoreSettings.toConfiguration(options.getSavepointRestoreSettings(), configuration);
return new ExecutionConfigAccessor(configuration);
}
- private static void parseJarURLToConfig(final String jarFile, final Configuration configuration) {
- if (jarFile == null) {
- return;
- }
-
- try {
- final URL jarUrl = new File(jarFile).getAbsoluteFile().toURI().toURL();
- final List<URL> jarUrlSingleton = Collections.singletonList(jarUrl);
- ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, jarUrlSingleton, URL::toString);
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException("JAR file path invalid", e);
- }
- }
-
public Configuration getConfiguration() {
return configuration;
}
- public String getJarFilePath() {
- final List<URL> jarURL = decodeUrlList(configuration, PipelineOptions.JARS);
- if (jarURL != null && !jarURL.isEmpty()) {
- return jarURL.get(0).getPath();
- }
- return null;
- }
-
public List<URL> getClasspaths() {
return decodeUrlList(configuration, PipelineOptions.CLASSPATHS);
}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
index 2b64c82..75771cc 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
@@ -80,11 +80,10 @@ public class CliFrontendPackageProgramTest extends TestLogger {
@Test
public void testNonExistingJarFile() throws Exception {
ProgramOptions programOptions = mock(ProgramOptions.class);
- ExecutionConfigAccessor executionOptions = mock(ExecutionConfigAccessor.class);
- when(executionOptions.getJarFilePath()).thenReturn("/some/none/existing/path");
+ when(programOptions.getJarFilePath()).thenReturn("/some/none/existing/path");
try {
- frontend.buildProgram(programOptions, executionOptions);
+ frontend.buildProgram(programOptions);
fail("should throw an exception");
}
catch (FileNotFoundException e) {
@@ -95,12 +94,11 @@ public class CliFrontendPackageProgramTest extends TestLogger {
@Test
public void testFileNotJarFile() throws Exception {
ProgramOptions programOptions = mock(ProgramOptions.class);
- ExecutionConfigAccessor executionOptions = mock(ExecutionConfigAccessor.class);
- when(executionOptions.getJarFilePath()).thenReturn(getNonJarFilePath());
+ when(programOptions.getJarFilePath()).thenReturn(getNonJarFilePath());
when(programOptions.getProgramArgs()).thenReturn(new String[0]);
try {
- frontend.buildProgram(programOptions, executionOptions);
+ frontend.buildProgram(programOptions);
fail("should throw an exception");
}
catch (ProgramInvocationException e) {
@@ -120,13 +118,12 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(getTestJarPath(), executionOptions.getJarFilePath());
- assertArrayEquals(classpath, executionOptions.getClasspaths().toArray());
+ assertEquals(getTestJarPath(), programOptions.getJarFilePath());
+ assertArrayEquals(classpath, programOptions.getClasspaths().toArray());
assertArrayEquals(reducedArguments, programOptions.getProgramArgs());
- PackagedProgram prog = frontend.buildProgram(programOptions, executionOptions);
+ PackagedProgram prog = frontend.buildProgram(programOptions);
Assert.assertArrayEquals(reducedArguments, prog.getArguments());
Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
@@ -144,13 +141,12 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(getTestJarPath(), executionOptions.getJarFilePath());
- assertArrayEquals(classpath, executionOptions.getClasspaths().toArray());
+ assertEquals(getTestJarPath(), programOptions.getJarFilePath());
+ assertArrayEquals(classpath, programOptions.getClasspaths().toArray());
assertArrayEquals(reducedArguments, programOptions.getProgramArgs());
- PackagedProgram prog = frontend.buildProgram(programOptions, executionOptions);
+ PackagedProgram prog = frontend.buildProgram(programOptions);
Assert.assertArrayEquals(reducedArguments, prog.getArguments());
Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
@@ -168,13 +164,12 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(getTestJarPath(), executionOptions.getJarFilePath());
- assertArrayEquals(classpath, executionOptions.getClasspaths().toArray());
+ assertEquals(getTestJarPath(), programOptions.getJarFilePath());
+ assertArrayEquals(classpath, programOptions.getClasspaths().toArray());
assertArrayEquals(reducedArguments, programOptions.getProgramArgs());
- PackagedProgram prog = frontend.buildProgram(programOptions, executionOptions);
+ PackagedProgram prog = frontend.buildProgram(programOptions);
Assert.assertArrayEquals(reducedArguments, prog.getArguments());
Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
@@ -197,14 +192,13 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(arguments[4], executionOptions.getJarFilePath());
- assertArrayEquals(classpath, executionOptions.getClasspaths().toArray());
+ assertEquals(arguments[4], programOptions.getJarFilePath());
+ assertArrayEquals(classpath, programOptions.getClasspaths().toArray());
assertArrayEquals(reducedArguments, programOptions.getProgramArgs());
try {
- frontend.buildProgram(programOptions, executionOptions);
+ frontend.buildProgram(programOptions);
fail("Should fail with an exception");
}
catch (FileNotFoundException e) {
@@ -218,13 +212,12 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(arguments[0], executionOptions.getJarFilePath());
+ assertEquals(arguments[0], programOptions.getJarFilePath());
assertArrayEquals(new String[0], programOptions.getProgramArgs());
try {
- frontend.buildProgram(programOptions, executionOptions);
+ frontend.buildProgram(programOptions);
}
catch (FileNotFoundException e) {
// that's what we want
@@ -279,14 +272,13 @@ public class CliFrontendPackageProgramTest extends TestLogger {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, arguments, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
- assertEquals(getTestJarPath(), executionOptions.getJarFilePath());
- assertArrayEquals(classpath, executionOptions.getClasspaths().toArray());
+ assertEquals(getTestJarPath(), programOptions.getJarFilePath());
+ assertArrayEquals(classpath, programOptions.getClasspaths().toArray());
assertEquals(TEST_JAR_CLASSLOADERTEST_CLASS, programOptions.getEntryPointClassName());
assertArrayEquals(reducedArguments, programOptions.getProgramArgs());
- PackagedProgram prog = spy(frontend.buildProgram(programOptions, executionOptions));
+ PackagedProgram prog = spy(frontend.buildProgram(programOptions));
ClassLoader testClassLoader = new ClassLoader(prog.getUserCodeClassLoader()) {
@Override
[flink] 05/07: [FLINK-14745] Add dependencies of job as list of
URLs in config
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bd8570f8551c78608b847109c0a61f8a46c6cb65
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Wed Nov 13 15:09:03 2019 +0100
[FLINK-14745] Add dependencies of job as list of URLs in config
---
.../src/main/java/org/apache/flink/client/ClientUtils.java | 2 +-
.../src/main/java/org/apache/flink/client/cli/CliFrontend.java | 3 ++-
.../org/apache/flink/client/cli/ExecutionConfigAccessor.java | 9 ++++++++-
.../java/org/apache/flink/client/program/PackagedProgram.java | 4 ++--
.../org/apache/flink/client/program/PackagedProgramUtils.java | 2 +-
.../java/org/apache/flink/client/cli/CliFrontendRunTest.java | 4 ++--
.../flink/table/client/gateway/local/ExecutionContext.java | 10 ++++------
7 files changed, 20 insertions(+), 14 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index f756449..043b740 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -147,7 +147,7 @@ public enum ClientUtils {
ContextEnvironmentFactory factory = new ContextEnvironmentFactory(
client,
- program.getAllLibraries(),
+ program.getJobJarAndDependencies(),
program.getClasspaths(),
program.getUserCodeClassLoader(),
parallelism,
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index 6e9b2f9..9ff6d53 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -211,7 +211,8 @@ public class CliFrontend {
final CustomCommandLine customCommandLine = getActiveCustomCommandLine(commandLine);
final Configuration executorConfig = customCommandLine.applyCommandLineOptionsToConfiguration(commandLine);
- final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromProgramOptions(programOptions);
+ final List<URL> jobJars = program.getJobJarAndDependencies();
+ final ExecutionConfigAccessor executionParameters = ExecutionConfigAccessor.fromProgramOptions(programOptions, jobJars);
final Configuration executionConfig = executionParameters.getConfiguration();
try {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
index ee32449..ec627ac 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
@@ -55,15 +55,18 @@ public class ExecutionConfigAccessor {
/**
* Creates an {@link ExecutionConfigAccessor} based on the provided {@link ProgramOptions} as provided by the user through the CLI.
*/
- public static ExecutionConfigAccessor fromProgramOptions(final ProgramOptions options) {
+ public static ExecutionConfigAccessor fromProgramOptions(final ProgramOptions options, final List<URL> jobJars) {
checkNotNull(options);
+ checkNotNull(jobJars);
final Configuration configuration = new Configuration();
+
configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, options.getParallelism());
configuration.setBoolean(DeploymentOptions.ATTACHED, !options.getDetachedMode());
configuration.setBoolean(DeploymentOptions.SHUTDOWN_IF_ATTACHED, options.isShutdownOnAttachedExit());
ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, options.getClasspaths(), URL::toString);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, jobJars, URL::toString);
SavepointRestoreSettings.toConfiguration(options.getSavepointRestoreSettings(), configuration);
@@ -74,6 +77,10 @@ public class ExecutionConfigAccessor {
return configuration;
}
+ public List<URL> getJars() {
+ return decodeUrlList(configuration, PipelineOptions.JARS);
+ }
+
public List<URL> getClasspaths() {
return decodeUrlList(configuration, PipelineOptions.CLASSPATHS);
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
index 2b593ca..1d96646 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
@@ -140,7 +140,7 @@ public class PackagedProgram {
// now that we have an entry point, we can extract the nested jar files (if any)
this.extractedTempLibraries = jarFileUrl == null ? Collections.emptyList() : extractContainedLibraries(jarFileUrl);
this.classpaths = classpaths;
- this.userCodeClassLoader = ClientUtils.buildUserCodeClassLoader(getAllLibraries(), classpaths, getClass().getClassLoader());
+ this.userCodeClassLoader = ClientUtils.buildUserCodeClassLoader(getJobJarAndDependencies(), classpaths, getClass().getClassLoader());
// load the entry point class
this.mainClass = loadMainClass(entryPointClassName, userCodeClassLoader);
@@ -227,7 +227,7 @@ public class PackagedProgram {
/**
* Returns all provided libraries needed to run the program.
*/
- public List<URL> getAllLibraries() {
+ public List<URL> getJobJarAndDependencies() {
List<URL> libs = new ArrayList<URL>(this.extractedTempLibraries.size() + 1);
if (jarFile != null) {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
index 00ac231..edf3617 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
@@ -63,7 +63,7 @@ public class PackagedProgramUtils {
if (jobID != null) {
jobGraph.setJobID(jobID);
}
- jobGraph.addJars(packagedProgram.getAllLibraries());
+ jobGraph.addJars(packagedProgram.getJobJarAndDependencies());
jobGraph.setClasspaths(packagedProgram.getClasspaths());
jobGraph.setSavepointRestoreSettings(packagedProgram.getSavepointSettings());
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
index d2aff28..449e1b2 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
@@ -85,7 +85,7 @@ public class CliFrontendRunTest extends CliFrontendTestBase {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, parameters, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
+ ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions, Collections.emptyList());
SavepointRestoreSettings savepointSettings = executionOptions.getSavepointRestoreSettings();
assertTrue(savepointSettings.restoreSavepoint());
@@ -99,7 +99,7 @@ public class CliFrontendRunTest extends CliFrontendTestBase {
CommandLine commandLine = CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, parameters, true);
ProgramOptions programOptions = new ProgramOptions(commandLine);
- ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions);
+ ExecutionConfigAccessor executionOptions = ExecutionConfigAccessor.fromProgramOptions(programOptions, Collections.emptyList());
SavepointRestoreSettings savepointSettings = executionOptions.getSavepointRestoreSettings();
assertTrue(savepointSettings.restoreSavepoint());
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
index e37635f..99c43a8 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
@@ -113,7 +113,6 @@ public class ExecutionContext<ClusterID> {
private final SessionContext sessionContext;
private final Environment mergedEnv;
- private final List<URL> dependencies;
private final ClassLoader classLoader;
private final Map<String, Module> modules;
private final Map<String, Catalog> catalogs;
@@ -136,7 +135,6 @@ public class ExecutionContext<ClusterID> {
Configuration flinkConfig, ClusterClientServiceLoader clusterClientServiceLoader, Options commandLineOptions, List<CustomCommandLine> availableCommandLines) throws FlinkException {
this.sessionContext = sessionContext.copy(); // create internal copy because session context is mutable
this.mergedEnv = Environment.merge(defaultEnvironment, sessionContext.getEnvironment());
- this.dependencies = dependencies;
this.flinkConfig = flinkConfig;
// create class loader
@@ -184,7 +182,7 @@ public class ExecutionContext<ClusterID> {
clusterClientFactory = serviceLoader.getClusterClientFactory(executorConfig);
checkState(clusterClientFactory != null);
- executionParameters = createExecutionParameterProvider(commandLine);
+ executionParameters = createExecutionParameterProvider(commandLine, dependencies);
clusterId = clusterClientFactory.getClusterId(executorConfig);
clusterSpec = clusterClientFactory.getClusterSpecification(executorConfig);
}
@@ -262,10 +260,10 @@ public class ExecutionContext<ClusterID> {
throw new SqlExecutionException("Could not find a matching deployment.");
}
- private static ExecutionConfigAccessor createExecutionParameterProvider(CommandLine commandLine) {
+ private static ExecutionConfigAccessor createExecutionParameterProvider(CommandLine commandLine, List<URL> jobJars) {
try {
final ProgramOptions programOptions = new ProgramOptions(commandLine);
- return ExecutionConfigAccessor.fromProgramOptions(programOptions);
+ return ExecutionConfigAccessor.fromProgramOptions(programOptions, jobJars);
} catch (CliArgsException e) {
throw new SqlExecutionException("Invalid deployment run options.", e);
}
@@ -487,7 +485,7 @@ public class ExecutionContext<ClusterID> {
flinkConfig,
parallelism);
- jobGraph.addJars(dependencies);
+ jobGraph.addJars(executionParameters.getJars());
jobGraph.setClasspaths(executionParameters.getClasspaths());
jobGraph.setSavepointRestoreSettings(executionParameters.getSavepointRestoreSettings());
[flink] 07/07: [FLINK-14745] Wire the configuration to the
ClientUtils.executeProgram
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 03c125c41f12c1f27fe61a320c5a04af200e4ec0
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Wed Nov 13 16:23:03 2019 +0100
[FLINK-14745] Wire the configuration to the ClientUtils.executeProgram
---
.../java/org/apache/flink/client/ClientUtils.java | 29 ++++++++++++++------
.../org/apache/flink/client/cli/CliFrontend.java | 12 +++-----
.../flink/client/cli/CliFrontendRunTest.java | 7 +++--
.../apache/flink/client/program/ClientTest.java | 32 ++++++++++++++++++----
4 files changed, 55 insertions(+), 25 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 043b740..02b73a9 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -20,6 +20,7 @@ package org.apache.flink.client;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.cli.ExecutionConfigAccessor;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ContextEnvironment;
import org.apache.flink.client.program.ContextEnvironmentFactory;
@@ -27,9 +28,11 @@ import org.apache.flink.client.program.DetachedJobExecutionResult;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.ProgramMissingJobException;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.util.ExceptionUtils;
@@ -133,13 +136,23 @@ public enum ClientUtils {
}
public static JobSubmissionResult executeProgram(
+ Configuration configuration,
ClusterClient<?> client,
- PackagedProgram program,
- int parallelism,
- boolean detached) throws ProgramMissingJobException, ProgramInvocationException {
+ PackagedProgram program) throws ProgramMissingJobException, ProgramInvocationException {
+
+ final ExecutionConfigAccessor executionConfigAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
+
+ final List<URL> jobJars = executionConfigAccessor.getJars();
+ final List<URL> classpaths = executionConfigAccessor.getClasspaths();
+ final SavepointRestoreSettings savepointSettings = executionConfigAccessor.getSavepointRestoreSettings();
+ final int parallelism = executionConfigAccessor.getParallelism();
+ final boolean detached = executionConfigAccessor.getDetachedMode();
+
final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+ final ClassLoader userCodeClassLoader = ClientUtils.buildUserCodeClassLoader(jobJars, classpaths, contextClassLoader);
+
try {
- Thread.currentThread().setContextClassLoader(program.getUserCodeClassLoader());
+ Thread.currentThread().setContextClassLoader(userCodeClassLoader);
LOG.info("Starting program (detached: {})", detached);
@@ -147,12 +160,12 @@ public enum ClientUtils {
ContextEnvironmentFactory factory = new ContextEnvironmentFactory(
client,
- program.getJobJarAndDependencies(),
- program.getClasspaths(),
- program.getUserCodeClassLoader(),
+ jobJars,
+ classpaths,
+ userCodeClassLoader,
parallelism,
detached,
- program.getSavepointSettings(),
+ savepointSettings,
jobExecutionResult);
ContextEnvironment.setAsContext(factory);
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index 9ff6d53..5d24aee 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -278,11 +278,8 @@ public class CliFrontend {
try {
int userParallelism = executionParameters.getParallelism();
LOG.debug("User parallelism is set to {}", userParallelism);
- if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
- userParallelism = defaultParallelism;
- }
- executeProgram(program, client, userParallelism, executionParameters.getDetachedMode());
+ executeProgram(executionConfig, program, client);
} finally {
if (clusterId == null && !executionParameters.getDetachedMode()) {
// terminate the cluster only if we have started it before and if it's not detached
@@ -742,13 +739,12 @@ public class CliFrontend {
// --------------------------------------------------------------------------------------------
protected void executeProgram(
+ Configuration configuration,
PackagedProgram program,
- ClusterClient<?> client,
- int parallelism,
- boolean detached) throws ProgramMissingJobException, ProgramInvocationException {
+ ClusterClient<?> client) throws ProgramMissingJobException, ProgramInvocationException {
logAndSysout("Starting execution of program");
- JobSubmissionResult result = ClientUtils.executeProgram(client, program, parallelism, detached);
+ JobSubmissionResult result = ClientUtils.executeProgram(configuration, client, program);
if (result.isJobExecutionResult()) {
logAndSysout("Program execution finished");
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
index 449e1b2..a0d551b 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
@@ -198,9 +198,10 @@ public class CliFrontendRunTest extends CliFrontendTestBase {
}
@Override
- protected void executeProgram(PackagedProgram program, ClusterClient client, int parallelism, boolean detached) {
- assertEquals(isDetached, detached);
- assertEquals(expectedParallelism, parallelism);
+ protected void executeProgram(Configuration configuration, PackagedProgram program, ClusterClient client) {
+ final ExecutionConfigAccessor executionConfigAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
+ assertEquals(isDetached, executionConfigAccessor.getDetachedMode());
+ assertEquals(expectedParallelism, executionConfigAccessor.getParallelism());
}
}
}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index a1cc8a2..5845080 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -31,8 +31,12 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.costs.DefaultCostEstimator;
@@ -52,6 +56,7 @@ import org.junit.experimental.categories.Category;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import java.net.URL;
import java.util.Collections;
import static org.junit.Assert.assertEquals;
@@ -92,6 +97,15 @@ public class ClientTest extends TestLogger {
config.setString(AkkaOptions.ASK_TIMEOUT, AkkaOptions.ASK_TIMEOUT.defaultValue());
}
+ private Configuration fromPackagedProgram(final PackagedProgram program, final int parallelism, final boolean detached) {
+ final Configuration configuration = new Configuration();
+ configuration.set(CoreOptions.DEFAULT_PARALLELISM, parallelism);
+ configuration.set(DeploymentOptions.ATTACHED, !detached);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, program.getClasspaths(), URL::toString);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, program.getJobJarAndDependencies(), URL::toString);
+ return configuration;
+ }
+
/**
* Tests that invalid detached mode programs fail.
*/
@@ -100,7 +114,8 @@ public class ClientTest extends TestLogger {
final ClusterClient<?> clusterClient = new MiniClusterClient(new Configuration(), MINI_CLUSTER_RESOURCE.getMiniCluster());
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestExecuteTwice.class.getName()).build();
- ClientUtils.executeProgram(clusterClient, prg, 1, true);
+ final Configuration configuration = fromPackagedProgram(prg, 1, true);
+ ClientUtils.executeProgram(configuration, clusterClient, prg);
fail(FAIL_MESSAGE);
} catch (ProgramInvocationException e) {
assertEquals(
@@ -110,7 +125,8 @@ public class ClientTest extends TestLogger {
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestEager.class.getName()).build();
- ClientUtils.executeProgram(clusterClient, prg, 1, true);
+ final Configuration configuration = fromPackagedProgram(prg, 1, true);
+ ClientUtils.executeProgram(configuration, clusterClient, prg);
fail(FAIL_MESSAGE);
} catch (ProgramInvocationException e) {
assertEquals(
@@ -120,7 +136,8 @@ public class ClientTest extends TestLogger {
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestGetRuntime.class.getName()).build();
- ClientUtils.executeProgram(clusterClient, prg, 1, true);
+ final Configuration configuration = fromPackagedProgram(prg, 1, true);
+ ClientUtils.executeProgram(configuration, clusterClient, prg);
fail(FAIL_MESSAGE);
} catch (ProgramInvocationException e) {
assertEquals(
@@ -130,7 +147,8 @@ public class ClientTest extends TestLogger {
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestGetAccumulator.class.getName()).build();
- ClientUtils.executeProgram(clusterClient, prg, 1, true);
+ final Configuration configuration = fromPackagedProgram(prg, 1, true);
+ ClientUtils.executeProgram(configuration, clusterClient, prg);
fail(FAIL_MESSAGE);
} catch (ProgramInvocationException e) {
assertEquals(
@@ -140,7 +158,8 @@ public class ClientTest extends TestLogger {
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestGetAllAccumulator.class.getName()).build();
- ClientUtils.executeProgram(clusterClient, prg, 1, true);
+ final Configuration configuration = fromPackagedProgram(prg, 1, true);
+ ClientUtils.executeProgram(configuration, clusterClient, prg);
fail(FAIL_MESSAGE);
} catch (ProgramInvocationException e) {
assertEquals(
@@ -184,7 +203,8 @@ public class ClientTest extends TestLogger {
try {
final ClusterClient<?> client = new MiniClusterClient(new Configuration(), MINI_CLUSTER_RESOURCE.getMiniCluster());
- ClientUtils.executeProgram(client, packagedProgramMock, 1, true);
+ final Configuration configuration = fromPackagedProgram(packagedProgramMock, 1, true);
+ ClientUtils.executeProgram(configuration, client, packagedProgramMock);
fail("Creating the local execution environment should not be possible");
}
catch (InvalidProgramException e) {
[flink] 03/07: [hotfix] Refactors
PackagedProgram.extractContainedLibraries()
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c2dbf94174d1f1cc08d86b4af9aabef4b28858bf
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Mon Nov 11 11:08:15 2019 +0100
[hotfix] Refactors PackagedProgram.extractContainedLibraries()
---
.../java/org/apache/flink/client/ClientUtils.java | 4 +-
.../flink/client/program/PackagedProgram.java | 140 +++++++++------------
2 files changed, 61 insertions(+), 83 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 6245002..f756449 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -143,13 +143,11 @@ public enum ClientUtils {
LOG.info("Starting program (detached: {})", detached);
- final List<URL> libraries = program.getAllLibraries();
-
final AtomicReference<JobExecutionResult> jobExecutionResult = new AtomicReference<>();
ContextEnvironmentFactory factory = new ContextEnvironmentFactory(
client,
- libraries,
+ program.getAllLibraries(),
program.getClasspaths(),
program.getUserCodeClassLoader(),
parallelism,
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
index 0a09367..2b593ca 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
@@ -46,13 +46,13 @@ import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Enumeration;
import java.util.List;
import java.util.Random;
import java.util.jar.Attributes;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.jar.Manifest;
+import java.util.stream.Collectors;
import static org.apache.flink.client.program.PackagedProgramUtils.isPython;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -427,100 +427,80 @@ public class PackagedProgram {
* @throws ProgramInvocationException Thrown, if the extraction process failed.
*/
public static List<File> extractContainedLibraries(URL jarFile) throws ProgramInvocationException {
+ try (final JarFile jar = new JarFile(new File(jarFile.toURI()))) {
- Random rnd = new Random();
-
- JarFile jar = null;
- try {
- jar = new JarFile(new File(jarFile.toURI()));
- final List<JarEntry> containedJarFileEntries = new ArrayList<JarEntry>();
-
- Enumeration<JarEntry> entries = jar.entries();
- while (entries.hasMoreElements()) {
- JarEntry entry = entries.nextElement();
- String name = entry.getName();
-
- if (name.length() > 8 && name.startsWith("lib/") && name.endsWith(".jar")) {
- containedJarFileEntries.add(entry);
- }
- }
-
+ final List<JarEntry> containedJarFileEntries = getContainedJarEntries(jar);
if (containedJarFileEntries.isEmpty()) {
return Collections.emptyList();
- } else {
- // go over all contained jar files
- final List<File> extractedTempLibraries = new ArrayList<File>(containedJarFileEntries.size());
- final byte[] buffer = new byte[4096];
+ }
- boolean incomplete = true;
-
- try {
- for (int i = 0; i < containedJarFileEntries.size(); i++) {
- final JarEntry entry = containedJarFileEntries.get(i);
- String name = entry.getName();
- // '/' as in case of zip, jar
- // java.util.zip.ZipEntry#isDirectory always looks only for '/' not for File.separator
- name = name.replace('/', '_');
-
- File tempFile;
- try {
- tempFile = File.createTempFile(rnd.nextInt(Integer.MAX_VALUE) + "_", name);
- tempFile.deleteOnExit();
- } catch (IOException e) {
- throw new ProgramInvocationException(
- "An I/O error occurred while creating temporary file to extract nested library '" +
- entry.getName() + "'.", e);
- }
+ final List<File> extractedTempLibraries = new ArrayList<>(containedJarFileEntries.size());
+ boolean incomplete = true;
- extractedTempLibraries.add(tempFile);
-
- // copy the temp file contents to a temporary File
- OutputStream out = null;
- InputStream in = null;
- try {
-
- out = new FileOutputStream(tempFile);
- in = new BufferedInputStream(jar.getInputStream(entry));
-
- int numRead = 0;
- while ((numRead = in.read(buffer)) != -1) {
- out.write(buffer, 0, numRead);
- }
- } catch (IOException e) {
- throw new ProgramInvocationException("An I/O error occurred while extracting nested library '"
- + entry.getName() + "' to temporary file '" + tempFile.getAbsolutePath() + "'.");
- } finally {
- if (out != null) {
- out.close();
- }
- if (in != null) {
- in.close();
- }
- }
- }
+ try {
+ final Random rnd = new Random();
+ final byte[] buffer = new byte[4096];
- incomplete = false;
- } finally {
- if (incomplete) {
- deleteExtractedLibraries(extractedTempLibraries);
- }
+ for (final JarEntry entry : containedJarFileEntries) {
+ // '/' as in case of zip, jar
+ // java.util.zip.ZipEntry#isDirectory always looks only for '/' not for File.separator
+ final String name = entry.getName().replace('/', '_');
+ final File tempFile = copyLibToTempFile(name, rnd, jar, entry, buffer);
+ extractedTempLibraries.add(tempFile);
}
- return extractedTempLibraries;
+ incomplete = false;
+ } finally {
+ if (incomplete) {
+ deleteExtractedLibraries(extractedTempLibraries);
+ }
}
+
+ return extractedTempLibraries;
} catch (Throwable t) {
throw new ProgramInvocationException("Unknown I/O error while extracting contained jar files.", t);
- } finally {
- if (jar != null) {
- try {
- jar.close();
- } catch (Throwable t) {
- }
+ }
+ }
+
+ private static File copyLibToTempFile(String name, Random rnd, JarFile jar, JarEntry input, byte[] buffer) throws ProgramInvocationException {
+ final File output = createTempFile(rnd, input, name);
+ try (
+ final OutputStream out = new FileOutputStream(output);
+ final InputStream in = new BufferedInputStream(jar.getInputStream(input))
+ ) {
+ int numRead = 0;
+ while ((numRead = in.read(buffer)) != -1) {
+ out.write(buffer, 0, numRead);
}
+ return output;
+ } catch (IOException e) {
+ throw new ProgramInvocationException("An I/O error occurred while extracting nested library '"
+ + input.getName() + "' to temporary file '" + output.getAbsolutePath() + "'.");
+ }
+ }
+
+ private static File createTempFile(Random rnd, JarEntry entry, String name) throws ProgramInvocationException {
+ try {
+ final File tempFile = File.createTempFile(rnd.nextInt(Integer.MAX_VALUE) + "_", name);
+ tempFile.deleteOnExit();
+ return tempFile;
+ } catch (IOException e) {
+ throw new ProgramInvocationException(
+ "An I/O error occurred while creating temporary file to extract nested library '" +
+ entry.getName() + "'.", e);
}
}
- public static void deleteExtractedLibraries(List<File> tempLibraries) {
+ private static List<JarEntry> getContainedJarEntries(JarFile jar) {
+ return jar.stream()
+ .filter(jarEntry -> {
+ final String name = jarEntry.getName();
+ return name.length() > 8 && name.startsWith("lib/") && name.endsWith(".jar");
+ })
+ .collect(Collectors.toList());
+ }
+
+ private static void deleteExtractedLibraries(List<File> tempLibraries) {
for (File f : tempLibraries) {
f.delete();
}
[flink] 06/07: [FLINK-14745] Fix setting
CoreOptions.DEFAULT_PARALLELISM in ExecutionConfigAccessor
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8293c2845955348882ca91beada413d52f233412
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Wed Nov 13 15:10:18 2019 +0100
[FLINK-14745] Fix setting CoreOptions.DEFAULT_PARALLELISM in ExecutionConfigAccessor
---
.../java/org/apache/flink/client/cli/ExecutionConfigAccessor.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
index ec627ac..9e570e1 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
@@ -19,6 +19,7 @@
package org.apache.flink.client.cli;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.configuration.Configuration;
@@ -61,7 +62,10 @@ public class ExecutionConfigAccessor {
final Configuration configuration = new Configuration();
- configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, options.getParallelism());
+ if (options.getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT) {
+ configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, options.getParallelism());
+ }
+
configuration.setBoolean(DeploymentOptions.ATTACHED, !options.getDetachedMode());
configuration.setBoolean(DeploymentOptions.SHUTDOWN_IF_ATTACHED, options.isShutdownOnAttachedExit());
[flink] 01/07: [hotfix] Change the signature of the
ConfigUtils.encodeCollectionToConfig()
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f6ac09fb42a1d8cda99500b23f7247f19a272516
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Wed Nov 13 14:33:54 2019 +0100
[hotfix] Change the signature of the ConfigUtils.encodeCollectionToConfig()
---
.../flink/client/cli/ExecutionConfigAccessor.java | 6 +-
.../apache/flink/configuration/ConfigUtils.java | 38 ++++---
.../flink/configuration/ConfigUtilsTest.java | 121 +++++++++++++++++++++
3 files changed, 148 insertions(+), 17 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
index 72a338a..77c0440 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutionConfigAccessor.java
@@ -65,9 +65,7 @@ public class ExecutionConfigAccessor {
configuration.setBoolean(DeploymentOptions.ATTACHED, !options.getDetachedMode());
configuration.setBoolean(DeploymentOptions.SHUTDOWN_IF_ATTACHED, options.isShutdownOnAttachedExit());
- if (options.getClasspaths() != null) {
- ConfigUtils.encodeStreamToConfig(configuration, PipelineOptions.CLASSPATHS, options.getClasspaths().stream(), URL::toString);
- }
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, options.getClasspaths(), URL::toString);
parseJarURLToConfig(options.getJarFilePath(), configuration);
@@ -84,7 +82,7 @@ public class ExecutionConfigAccessor {
try {
final URL jarUrl = new File(jarFile).getAbsoluteFile().toURI().toURL();
final List<URL> jarUrlSingleton = Collections.singletonList(jarUrl);
- ConfigUtils.encodeStreamToConfig(configuration, PipelineOptions.JARS, jarUrlSingleton.stream(), URL::toString);
+ ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, jarUrlSingleton, URL::toString);
} catch (MalformedURLException e) {
throw new IllegalArgumentException("JAR file path invalid", e);
}
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java
index be95041..4b41389 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java
@@ -20,13 +20,16 @@ package org.apache.flink.configuration;
import org.apache.flink.annotation.Internal;
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -38,37 +41,45 @@ public class ConfigUtils {
/**
* Puts an array of values of type {@code IN} in a {@link WritableConfig}
- * as a {@link ConfigOption} of type {@link List} of type {@code OUT}.
+ * as a {@link ConfigOption} of type {@link List} of type {@code OUT}. If the {@code values}
+ * is {@code null} or empty, then nothing is put in the configuration.
*
* @param configuration the configuration object to put the list in
* @param key the {@link ConfigOption option} to serve as the key for the list in the configuration
- * @param value the array of values to put as value for the {@code key}
+ * @param values the array of values to put as value for the {@code key}
* @param mapper the transformation function from {@code IN} to {@code OUT}.
*/
public static <IN, OUT> void encodeArrayToConfig(
final WritableConfig configuration,
final ConfigOption<List<OUT>> key,
- final IN[] value,
+ @Nullable final IN[] values,
final Function<IN, OUT> mapper) {
- if (value == null) {
+
+ checkNotNull(configuration);
+ checkNotNull(key);
+ checkNotNull(mapper);
+
+ if (values == null) {
return;
}
- encodeStreamToConfig(configuration, key, Arrays.stream(value), mapper);
+
+ encodeCollectionToConfig(configuration, key, Arrays.asList(values), mapper);
}
/**
- * Puts a {@link Stream} of values of type {@code IN} in a {@link WritableConfig}
- * as a {@link ConfigOption} of type {@link List} of type {@code OUT}.
+ * Puts a {@link Collection} of values of type {@code IN} in a {@link WritableConfig}
+ * as a {@link ConfigOption} of type {@link List} of type {@code OUT}. If the {@code values}
+ * is {@code null} or empty, then nothing is put in the configuration.
*
* @param configuration the configuration object to put the list in
* @param key the {@link ConfigOption option} to serve as the key for the list in the configuration
- * @param values the stream of values to put as value for the {@code key}
+ * @param values the collection of values to put as value for the {@code key}
* @param mapper the transformation function from {@code IN} to {@code OUT}.
*/
- public static <IN, OUT> void encodeStreamToConfig(
+ public static <IN, OUT> void encodeCollectionToConfig(
final WritableConfig configuration,
final ConfigOption<List<OUT>> key,
- final Stream<IN> values,
+ @Nullable final Collection<IN> values,
final Function<IN, OUT> mapper) {
checkNotNull(configuration);
@@ -79,10 +90,11 @@ public class ConfigUtils {
return;
}
- final List<OUT> encodedOption = values
+ final List<OUT> encodedOption = values.stream()
+ .filter(Objects::nonNull)
.map(mapper)
.filter(Objects::nonNull)
- .collect(Collectors.toList());
+ .collect(Collectors.toCollection(ArrayList::new));
if (!encodedOption.isEmpty()) {
configuration.set(key, encodedOption);
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java
new file mode 100644
index 0000000..2edbd9f
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.configuration;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests the {@link ConfigUtils} methods.
+ */
+public class ConfigUtilsTest {
+
+ private static final ConfigOption<List<String>> TEST_OPTION =
+ key("test.option.key")
+ .stringType()
+ .asList()
+ .noDefaultValue();
+
+ private static final Integer[] intArray = {1, 3, 2, 4};
+ private static final List<Integer> intList = Arrays.asList(intArray);
+
+ @Test
+ public void collectionIsCorrectlyPutAndFetched() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeCollectionToConfig(configurationUnderTest, TEST_OPTION, intList, Object::toString);
+
+ final List<Integer> recovered = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recovered, equalTo(intList));
+ }
+
+ @Test
+ public void arrayIsCorrectlyPutAndFetched() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeArrayToConfig(configurationUnderTest, TEST_OPTION, intArray, Object::toString);
+
+ final List<Integer> recovered = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recovered, equalTo(intList));
+ }
+
+ @Test
+ public void nullCollectionPutsNothingInConfig() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeCollectionToConfig(configurationUnderTest, TEST_OPTION, null, Object::toString);
+
+ assertThat(configurationUnderTest.keySet(), is(empty()));
+
+ final Object recovered = configurationUnderTest.get(TEST_OPTION);
+ assertThat(recovered, is(nullValue()));
+
+ final List<Integer> recoveredList = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recoveredList, is(empty()));
+ }
+
+ @Test
+ public void nullArrayPutsNothingInConfig() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeArrayToConfig(configurationUnderTest, TEST_OPTION, null, Object::toString);
+
+ assertThat(configurationUnderTest.keySet(), is(empty()));
+
+ final Object recovered = configurationUnderTest.get(TEST_OPTION);
+ assertThat(recovered, is(nullValue()));
+
+ final List<Integer> recoveredList = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recoveredList, is(empty()));
+ }
+
+ @Test
+ public void emptyCollectionPutsNothingInConfig() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeCollectionToConfig(configurationUnderTest, TEST_OPTION, Collections.emptyList(), Object::toString);
+
+ assertThat(configurationUnderTest.keySet(), is(empty()));
+
+ final Object recovered = configurationUnderTest.get(TEST_OPTION);
+ assertThat(recovered, is(nullValue()));
+
+ final List<Integer> recoveredList = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recoveredList, is(empty()));
+ }
+
+ @Test
+ public void emptyArrayPutsNothingInConfig() {
+ final Configuration configurationUnderTest = new Configuration();
+ ConfigUtils.encodeArrayToConfig(configurationUnderTest, TEST_OPTION, new Integer[5], Object::toString);
+
+ assertThat(configurationUnderTest.keySet(), is(empty()));
+
+ final Object recovered = configurationUnderTest.get(TEST_OPTION);
+ assertThat(recovered, is(nullValue()));
+
+ final List<Integer> recoveredList = ConfigUtils.decodeListFromConfig(configurationUnderTest, TEST_OPTION, Integer::valueOf);
+ assertThat(recoveredList, is(empty()));
+ }
+}
[flink] 02/07: [hotfix] Move addJars() from ClientUtils to JobGraph
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 86b00ad0fa5ef998179f22aee62efdd635945872
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Mon Nov 11 11:15:37 2019 +0100
[hotfix] Move addJars() from ClientUtils to JobGraph
---
.../main/java/org/apache/flink/client/ClientUtils.java | 15 ---------------
.../java/org/apache/flink/client/RemoteExecutor.java | 2 +-
.../apache/flink/client/program/ContextEnvironment.java | 2 +-
.../flink/client/program/PackagedProgramUtils.java | 3 +--
.../org/apache/flink/client/program/ClientTest.java | 2 +-
.../org/apache/flink/runtime/jobgraph/JobGraph.java | 17 +++++++++++++++++
.../api/environment/StreamContextEnvironment.java | 2 +-
.../table/client/gateway/local/ExecutionContext.java | 3 +--
8 files changed, 23 insertions(+), 23 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 176d586..6245002 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -27,7 +27,6 @@ import org.apache.flink.client.program.DetachedJobExecutionResult;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.ProgramMissingJobException;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -56,20 +55,6 @@ public enum ClientUtils {
private static final Logger LOG = LoggerFactory.getLogger(ClientUtils.class);
- /**
- * Adds the given jar files to the {@link JobGraph} via {@link JobGraph#addJar}. This will
- * throw an exception if a jar URL is not valid.
- */
- public static void addJarFiles(JobGraph jobGraph, List<URL> jarFilesToAttach) {
- for (URL jar : jarFilesToAttach) {
- try {
- jobGraph.addJar(new Path(jar.toURI()));
- } catch (URISyntaxException e) {
- throw new RuntimeException("URL is invalid. This should not happen.", e);
- }
- }
- }
-
public static void checkJarFile(URL jar) throws IOException {
File jarFile;
try {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java
index b601367..71fde64 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java
@@ -108,7 +108,7 @@ public class RemoteExecutor extends PlanExecutor {
clientConfiguration,
getDefaultParallelism());
- ClientUtils.addJarFiles(jobGraph, jarFiles);
+ jobGraph.addJars(jarFiles);
jobGraph.setClasspaths(globalClasspaths);
ClassLoader userCodeClassLoader = ClientUtils.buildUserCodeClassLoader(
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 8f9048c..08a02af 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
@@ -84,7 +84,7 @@ public class ContextEnvironment extends ExecutionEnvironment {
client.getFlinkConfiguration(),
getParallelism());
- ClientUtils.addJarFiles(jobGraph, this.jarFilesToAttach);
+ jobGraph.addJars(this.jarFilesToAttach);
jobGraph.setClasspaths(this.classpathsToAttach);
if (detached) {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
index 2f27193..00ac231 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
@@ -20,7 +20,6 @@ package org.apache.flink.client.program;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.dag.Pipeline;
-import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.optimizer.CompilerException;
@@ -64,7 +63,7 @@ public class PackagedProgramUtils {
if (jobID != null) {
jobGraph.setJobID(jobID);
}
- ClientUtils.addJarFiles(jobGraph, packagedProgram.getAllLibraries());
+ jobGraph.addJars(packagedProgram.getAllLibraries());
jobGraph.setClasspaths(packagedProgram.getClasspaths());
jobGraph.setSavepointRestoreSettings(packagedProgram.getSavepointSettings());
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index 78023d7..a1cc8a2 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -160,7 +160,7 @@ public class ClientTest extends TestLogger {
new Configuration(),
1);
- ClientUtils.addJarFiles(jobGraph, Collections.emptyList());
+ jobGraph.addJars(Collections.emptyList());
jobGraph.setClasspaths(Collections.emptyList());
JobSubmissionResult result = ClientUtils.submitJob(clusterClient, jobGraph);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index f63a518..dceb517 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -31,6 +31,7 @@ import org.apache.flink.util.SerializedValue;
import java.io.IOException;
import java.io.Serializable;
+import java.net.URISyntaxException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
@@ -474,6 +475,22 @@ public class JobGraph implements Serializable {
}
/**
+ * Adds the given jar files to the {@link JobGraph} via {@link JobGraph#addJar}.
+ *
+ * @param jarFilesToAttach a list of the {@link URL URLs} of the jar files to attach to the jobgraph.
+ * @throws RuntimeException if a jar URL is not valid.
+ */
+ public void addJars(final List<URL> jarFilesToAttach) {
+ for (URL jar : jarFilesToAttach) {
+ try {
+ addJar(new Path(jar.toURI()));
+ } catch (URISyntaxException e) {
+ throw new RuntimeException("URL is invalid. This should not happen.", e);
+ }
+ }
+ }
+
+ /**
* Gets the list of assigned user jar paths.
*
* @return The list of assigned user jar paths
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 278a75b..bab31d3 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
@@ -51,7 +51,7 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
ctx.getClient().getFlinkConfiguration(),
getParallelism());
- ClientUtils.addJarFiles(jobGraph, ctx.getJars());
+ jobGraph.addJars(ctx.getJars());
jobGraph.setClasspaths(ctx.getClasspaths());
// running from the CLI will override the savepoint restore settings
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
index 692ebee..e37635f 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.client.cli.CliArgsException;
import org.apache.flink.client.cli.CustomCommandLine;
@@ -488,7 +487,7 @@ public class ExecutionContext<ClusterID> {
flinkConfig,
parallelism);
- ClientUtils.addJarFiles(jobGraph, dependencies);
+ jobGraph.addJars(dependencies);
jobGraph.setClasspaths(executionParameters.getClasspaths());
jobGraph.setSavepointRestoreSettings(executionParameters.getSavepointRestoreSettings());