You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/01/12 22:25:24 UTC
[03/14] flink git commit: [FLINK-8340] [flip6] Remove passing of
Configuration to CustomCommandLine
[FLINK-8340] [flip6] Remove passing of Configuration to CustomCommandLine
Since the Configuration does not change over the lifetime of a CustomCommandLine,
we can safely pass it as a constructor argument instead of method argument.
This closes #5226.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/30011b9b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/30011b9b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/30011b9b
Branch: refs/heads/master
Commit: 30011b9b110aad0e1c28e7e0a025b73986781a72
Parents: 4ade826
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Jan 3 13:25:09 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Fri Jan 12 16:14:02 2018 +0100
----------------------------------------------------------------------
.../client/cli/AbstractCustomCommandLine.java | 16 +-
.../apache/flink/client/cli/CliFrontend.java | 84 ++--
.../flink/client/cli/CustomCommandLine.java | 23 +-
.../org/apache/flink/client/cli/DefaultCLI.java | 17 +-
.../flink/client/cli/Flip6DefaultCLI.java | 17 +-
.../flink/client/cli/CliFrontendCancelTest.java | 24 +-
.../flink/client/cli/CliFrontendInfoTest.java | 24 +-
.../flink/client/cli/CliFrontendListTest.java | 6 +-
.../cli/CliFrontendPackageProgramTest.java | 251 +++++-------
.../flink/client/cli/CliFrontendRunTest.java | 34 +-
.../flink/client/cli/CliFrontendStopTest.java | 12 +-
.../apache/flink/client/cli/DefaultCLITest.java | 49 +--
.../flink/client/cli/Flip6DefaultCLITest.java | 4 +-
.../client/cli/util/DummyCustomCommandLine.java | 12 +-
.../client/cli/util/MockedCliFrontend.java | 4 +-
.../flink/client/program/ClusterClientTest.java | 4 +
.../org/apache/flink/api/scala/FlinkShell.scala | 27 +-
.../YARNSessionCapacitySchedulerITCase.java | 4 +-
.../flink/yarn/YARNSessionFIFOITCase.java | 5 +-
.../org/apache/flink/yarn/YarnTestBase.java | 5 +-
.../apache/flink/yarn/YarnClusterClient.java | 3 +-
.../flink/yarn/cli/FlinkYarnSessionCli.java | 389 ++++++++++---------
.../flink/yarn/FlinkYarnSessionCliTest.java | 121 +++---
23 files changed, 541 insertions(+), 594 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java b/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java
index 70ffc5c..c7a1672 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java
@@ -22,6 +22,9 @@ import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.Preconditions;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
@@ -47,6 +50,16 @@ public abstract class AbstractCustomCommandLine<C extends ClusterClient> impleme
"Address of the JobManager (master) to which to connect. " +
"Use this flag to connect to a different JobManager than the one specified in the configuration.");
+ protected final Configuration configuration;
+
+ protected AbstractCustomCommandLine(Configuration configuration) {
+ this.configuration = new UnmodifiableConfiguration(Preconditions.checkNotNull(configuration));
+ }
+
+ public Configuration getConfiguration() {
+ return configuration;
+ }
+
@Override
public void addRunOptions(Options baseOptions) {
// nothing to add here
@@ -61,11 +74,10 @@ public abstract class AbstractCustomCommandLine<C extends ClusterClient> impleme
/**
* Override configuration settings by specified command line options.
*
- * @param configuration to use as the base configuration
* @param commandLine containing the overriding values
* @return Effective configuration with the overriden configuration settings
*/
- protected Configuration applyCommandLineOptionsToConfiguration(Configuration configuration, CommandLine commandLine) {
+ protected Configuration applyCommandLineOptionsToConfiguration(CommandLine commandLine) throws FlinkException {
final Configuration resultingConfiguration = new Configuration(configuration);
if (commandLine.hasOption(addressOption.getOpt())) {
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
----------------------------------------------------------------------
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 4b5040e..630154c 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
@@ -108,8 +108,6 @@ public class CliFrontend {
private final List<CustomCommandLine<?>> customCommandLines;
- private final String configurationDirectory;
-
private final Options customCommandLineOptions;
private final FiniteDuration clientTimeout;
@@ -118,11 +116,9 @@ public class CliFrontend {
public CliFrontend(
Configuration configuration,
- List<CustomCommandLine<?>> customCommandLines,
- String configurationDirectory) throws Exception {
+ List<CustomCommandLine<?>> customCommandLines) throws Exception {
this.configuration = Preconditions.checkNotNull(configuration);
this.customCommandLines = Preconditions.checkNotNull(customCommandLines);
- this.configurationDirectory = Preconditions.checkNotNull(configurationDirectory);
try {
FileSystem.initialize(this.configuration);
@@ -139,9 +135,9 @@ public class CliFrontend {
}
this.clientTimeout = AkkaUtils.getClientTimeout(this.configuration);
- this.defaultParallelism = GlobalConfiguration.loadConfiguration().getInteger(
- ConfigConstants.DEFAULT_PARALLELISM_KEY,
- ConfigConstants.DEFAULT_PARALLELISM);
+ this.defaultParallelism = configuration.getInteger(
+ ConfigConstants.DEFAULT_PARALLELISM_KEY,
+ ConfigConstants.DEFAULT_PARALLELISM);
}
// --------------------------------------------------------------------------------------------
@@ -161,15 +157,6 @@ public class CliFrontend {
return copiedConfiguration;
}
- /**
- * Returns the configuration directory for the CLI frontend.
- *
- * @return Configuration directory
- */
- public String getConfigurationDirectory() {
- return configurationDirectory;
- }
-
// --------------------------------------------------------------------------------------------
// Execute Actions
// --------------------------------------------------------------------------------------------
@@ -211,20 +198,17 @@ public class CliFrontend {
final CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(commandLine);
try {
- final String clusterId = customCommandLine.getClusterId(configuration, commandLine);
+ final String clusterId = customCommandLine.getClusterId(commandLine);
final ClusterClient client;
if (clusterId != null) {
client = clusterDescriptor.retrieve(clusterId);
} else {
- final ClusterSpecification clusterSpecification = customCommandLine.getClusterSpecification(configuration, commandLine);
+ final ClusterSpecification clusterSpecification = customCommandLine.getClusterSpecification(commandLine);
client = clusterDescriptor.deploySessionCluster(clusterSpecification);
}
@@ -368,14 +352,9 @@ public class CliFrontend {
}
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(commandLine);
- final String clusterId = activeCommandLine.getClusterId(
- configuration,
- commandLine);
+ final String clusterId = activeCommandLine.getClusterId(commandLine);
if (clusterId == null) {
throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
@@ -496,12 +475,9 @@ public class CliFrontend {
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(commandLine);
- final String clusterId = activeCommandLine.getClusterId(configuration, commandLine);
+ final String clusterId = activeCommandLine.getClusterId(commandLine);
if (clusterId == null) {
throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
@@ -579,12 +555,9 @@ public class CliFrontend {
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(commandLine);
- final String clusterId = activeCommandLine.getClusterId(configuration, commandLine);
+ final String clusterId = activeCommandLine.getClusterId(commandLine);
if (clusterId == null) {
throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
@@ -646,14 +619,9 @@ public class CliFrontend {
CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(commandLine);
- final String clusterId = customCommandLine.getClusterId(
- configuration,
- commandLine);
+ final String clusterId = customCommandLine.getClusterId(commandLine);
if (clusterId == null) {
throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
@@ -1004,13 +972,14 @@ public class CliFrontend {
final Configuration configuration = GlobalConfiguration.loadConfiguration(configurationDirectory);
// 3. load the custom command lines
- final List<CustomCommandLine<?>> customCommandLines = loadCustomCommandLines();
+ final List<CustomCommandLine<?>> customCommandLines = loadCustomCommandLines(
+ configuration,
+ configurationDirectory);
try {
final CliFrontend cli = new CliFrontend(
configuration,
- customCommandLines,
- configurationDirectory);
+ customCommandLines);
SecurityUtils.install(new SecurityConfiguration(cli.configuration));
int retCode = SecurityUtils.getInstalledContext()
@@ -1070,7 +1039,7 @@ public class CliFrontend {
config.setInteger(JobManagerOptions.PORT, address.getPort());
}
- public static List<CustomCommandLine<?>> loadCustomCommandLines() {
+ public static List<CustomCommandLine<?>> loadCustomCommandLines(Configuration configuration, String configurationDirectory) {
List<CustomCommandLine<?>> customCommandLines = new ArrayList<>(2);
// Command line interface of the YARN session, with a special initialization here
@@ -1079,13 +1048,18 @@ public class CliFrontend {
// active CustomCommandLine in order and DefaultCLI isActive always return true.
final String flinkYarnSessionCLI = "org.apache.flink.yarn.cli.FlinkYarnSessionCli";
try {
- customCommandLines.add(loadCustomCommandLine(flinkYarnSessionCLI, "y", "yarn"));
+ customCommandLines.add(
+ loadCustomCommandLine(flinkYarnSessionCLI,
+ configuration,
+ configurationDirectory,
+ "y",
+ "yarn"));
} catch (Exception e) {
LOG.warn("Could not load CLI class {}.", flinkYarnSessionCLI, e);
}
- customCommandLines.add(new Flip6DefaultCLI());
- customCommandLines.add(new DefaultCLI());
+ customCommandLines.add(new Flip6DefaultCLI(configuration));
+ customCommandLines.add(new DefaultCLI(configuration));
return customCommandLines;
}
@@ -1101,7 +1075,7 @@ public class CliFrontend {
*/
public CustomCommandLine getActiveCustomCommandLine(CommandLine commandLine) {
for (CustomCommandLine cli : customCommandLines) {
- if (cli.isActive(commandLine, configuration)) {
+ if (cli.isActive(commandLine)) {
return cli;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java
index cfa0827..f642484 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java
@@ -21,7 +21,7 @@ package org.apache.flink.client.cli;
import org.apache.flink.client.deployment.ClusterDescriptor;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.FlinkException;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
@@ -36,10 +36,9 @@ public interface CustomCommandLine<ClusterType extends ClusterClient> {
/**
* Signals whether the custom command-line wants to execute or not.
* @param commandLine The command-line options
- * @param configuration The Flink configuration
* @return True if the command-line wants to run, False otherwise
*/
- boolean isActive(CommandLine commandLine, Configuration configuration);
+ boolean isActive(CommandLine commandLine);
/**
* Gets the unique identifier of this CustomCommandLine.
@@ -55,6 +54,7 @@ public interface CustomCommandLine<ClusterType extends ClusterClient> {
/**
* Adds custom options to the existing general options.
+ *
* @param baseOptions The existing options.
*/
void addGeneralOptions(Options baseOptions);
@@ -63,15 +63,11 @@ public interface CustomCommandLine<ClusterType extends ClusterClient> {
* Create a {@link ClusterDescriptor} from the given configuration, configuration directory
* and the command line.
*
- * @param configuration to create the ClusterDescriptor with
- * @param configurationDirectory where the configuration was loaded from
* @param commandLine containing command line options relevant for the ClusterDescriptor
* @return ClusterDescriptor
+ * @throws FlinkException if the ClusterDescriptor could not be created
*/
- ClusterDescriptor<ClusterType> createClusterDescriptor(
- Configuration configuration,
- String configurationDirectory,
- CommandLine commandLine);
+ ClusterDescriptor<ClusterType> createClusterDescriptor(CommandLine commandLine) throws FlinkException;
/**
* Returns the cluster id if a cluster id was specified on the command line, otherwise it
@@ -80,24 +76,21 @@ public interface CustomCommandLine<ClusterType extends ClusterClient> {
* <p>A cluster id identifies a running cluster, e.g. the Yarn application id for a Flink
* cluster running on Yarn.
*
- * @param configuration to be used for the cluster id retrieval
* @param commandLine containing command line options relevant for the cluster id retrieval
* @return Cluster id identifying the cluster to deploy jobs to or null
*/
@Nullable
- String getClusterId(Configuration configuration, CommandLine commandLine);
+ String getClusterId(CommandLine commandLine);
/**
* Returns the {@link ClusterSpecification} specified by the configuration and the command
* line options. This specification can be used to deploy a new Flink cluster.
*
- * @param configuration to be used for the ClusterSpecification values
* @param commandLine containing command line options relevant for the ClusterSpecification
* @return ClusterSpecification for a new Flink cluster
+ * @throws FlinkException if the ClusterSpecification could not be created
*/
- ClusterSpecification getClusterSpecification(
- Configuration configuration,
- CommandLine commandLine);
+ ClusterSpecification getClusterSpecification(CommandLine commandLine) throws FlinkException;
default CommandLine parseCommandLineOptions(String[] args, boolean stopAtNonOptions) throws CliArgsException {
final Options options = new Options();
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java b/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
index d34e307..c29c5b7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/DefaultCLI.java
@@ -23,6 +23,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.StandaloneClusterDescriptor;
import org.apache.flink.client.program.StandaloneClusterClient;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.FlinkException;
import org.apache.commons.cli.CommandLine;
@@ -33,8 +34,12 @@ import javax.annotation.Nullable;
*/
public class DefaultCLI extends AbstractCustomCommandLine<StandaloneClusterClient> {
+ public DefaultCLI(Configuration configuration) {
+ super(configuration);
+ }
+
@Override
- public boolean isActive(CommandLine commandLine, Configuration configuration) {
+ public boolean isActive(CommandLine commandLine) {
// always active because we can try to read a JobManager address from the config
return true;
}
@@ -46,22 +51,20 @@ public class DefaultCLI extends AbstractCustomCommandLine<StandaloneClusterClien
@Override
public ClusterDescriptor<StandaloneClusterClient> createClusterDescriptor(
- Configuration configuration,
- String configurationDirectory,
- CommandLine commandLine) {
- final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(configuration, commandLine);
+ CommandLine commandLine) throws FlinkException {
+ final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(commandLine);
return new StandaloneClusterDescriptor(effectiveConfiguration);
}
@Override
@Nullable
- public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ public String getClusterId(CommandLine commandLine) {
return "standalone";
}
@Override
- public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ public ClusterSpecification getClusterSpecification(CommandLine commandLine) {
return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/main/java/org/apache/flink/client/cli/Flip6DefaultCLI.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/Flip6DefaultCLI.java b/flink-clients/src/main/java/org/apache/flink/client/cli/Flip6DefaultCLI.java
index 0e80f44..3adeca6 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/Flip6DefaultCLI.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/Flip6DefaultCLI.java
@@ -23,6 +23,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.Flip6StandaloneClusterDescriptor;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.FlinkException;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
@@ -41,8 +42,12 @@ public class Flip6DefaultCLI extends AbstractCustomCommandLine<RestClusterClient
FLIP_6.setRequired(false);
}
+ public Flip6DefaultCLI(Configuration configuration) {
+ super(configuration);
+ }
+
@Override
- public boolean isActive(CommandLine commandLine, Configuration configuration) {
+ public boolean isActive(CommandLine commandLine) {
return commandLine.hasOption(FLIP_6.getOpt());
}
@@ -59,22 +64,20 @@ public class Flip6DefaultCLI extends AbstractCustomCommandLine<RestClusterClient
@Override
public ClusterDescriptor<RestClusterClient> createClusterDescriptor(
- Configuration configuration,
- String configurationDirectory,
- CommandLine commandLine) {
- final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(configuration, commandLine);
+ CommandLine commandLine) throws FlinkException {
+ final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(commandLine);
return new Flip6StandaloneClusterDescriptor(effectiveConfiguration);
}
@Override
@Nullable
- public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ public String getClusterId(CommandLine commandLine) {
return "flip6Standalone";
}
@Override
- public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ public ClusterSpecification getClusterSpecification(CommandLine commandLine) {
return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendCancelTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendCancelTest.java
index b0a308d..60bd308 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendCancelTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendCancelTest.java
@@ -63,20 +63,20 @@ public class CliFrontendCancelTest extends TestLogger {
@Test(expected = CliArgsException.class)
public void testMissingJobId() throws Exception {
String[] parameters = {};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
@Test(expected = CliArgsException.class)
public void testUnrecognizedOption() throws Exception {
String[] parameters = {"-v", "-l"};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
@@ -116,10 +116,10 @@ public class CliFrontendCancelTest extends TestLogger {
public void testCancelWithSavepointWithoutJobId() throws Exception {
// Cancel with savepoint (with target directory), but no job ID
String[] parameters = { "-s", "targetDirectory" };
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
@@ -127,10 +127,10 @@ public class CliFrontendCancelTest extends TestLogger {
public void testCancelWithSavepointWithoutParameters() throws Exception {
// Cancel with savepoint (no target directory) and no job ID
String[] parameters = { "-s" };
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendInfoTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendInfoTest.java
index fa212cb..c284c61 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendInfoTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendInfoTest.java
@@ -42,20 +42,20 @@ public class CliFrontendInfoTest extends TestLogger {
@Test(expected = CliArgsException.class)
public void testMissingOption() throws Exception {
String[] parameters = {};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
@Test(expected = CliArgsException.class)
public void testUnrecognizedOption() throws Exception {
String[] parameters = {"-v", "-l"};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.cancel(parameters);
}
@@ -65,10 +65,10 @@ public class CliFrontendInfoTest extends TestLogger {
try {
String[] parameters = new String[]{CliFrontendTestUtils.getTestJarPath(), "-f", "true"};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.info(parameters);
assertTrue(buffer.toString().contains("\"parallelism\": \"1\""));
}
@@ -82,10 +82,10 @@ public class CliFrontendInfoTest extends TestLogger {
replaceStdOut();
try {
String[] parameters = {"-p", "17", CliFrontendTestUtils.getTestJarPath()};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.info(parameters);
assertTrue(buffer.toString().contains("\"parallelism\": \"17\""));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
index 6639e25..77d8016 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendListTest.java
@@ -60,10 +60,10 @@ public class CliFrontendListTest extends TestLogger {
@Test(expected = CliArgsException.class)
public void testUnrecognizedOption() throws Exception {
String[] parameters = {"-v", "-k"};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.list(parameters);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
----------------------------------------------------------------------
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 a862dc1..6873e68 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
@@ -22,7 +22,6 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.optimizer.CompilerException;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.costs.DefaultCostEstimator;
@@ -64,189 +63,145 @@ public class CliFrontendPackageProgramTest extends TestLogger {
@Before
public void setup() throws Exception {
+ final Configuration configuration = new Configuration();
frontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
}
@Test
- public void testNonExistingJarFile() {
- try {
- ProgramOptions options = mock(ProgramOptions.class);
- when(options.getJarFilePath()).thenReturn("/some/none/existing/path");
+ public void testNonExistingJarFile() throws Exception {
+ ProgramOptions options = mock(ProgramOptions.class);
+ when(options.getJarFilePath()).thenReturn("/some/none/existing/path");
- try {
- frontend.buildProgram(options);
- fail("should throw an exception");
- }
- catch (FileNotFoundException e) {
- // that's what we want
- }
+ try {
+ frontend.buildProgram(options);
+ fail("should throw an exception");
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+ catch (FileNotFoundException e) {
+ // that's what we want
}
}
@Test
- public void testFileNotJarFile() {
- try {
- ProgramOptions options = mock(ProgramOptions.class);
- when(options.getJarFilePath()).thenReturn(getNonJarFilePath());
+ public void testFileNotJarFile() throws Exception {
+ ProgramOptions options = mock(ProgramOptions.class);
+ when(options.getJarFilePath()).thenReturn(getNonJarFilePath());
- try {
- frontend.buildProgram(options);
- fail("should throw an exception");
- }
- catch (ProgramInvocationException e) {
- // that's what we want
- }
+ try {
+ frontend.buildProgram(options);
+ fail("should throw an exception");
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+ catch (ProgramInvocationException e) {
+ // that's what we want
}
}
@Test
- public void testVariantWithExplicitJarAndArgumentsOption() {
- try {
- String[] arguments = {
- "--classpath", "file:///tmp/foo",
- "--classpath", "file:///tmp/bar",
- "-j", getTestJarPath(),
- "-a", "--debug", "true", "arg1", "arg2" };
- URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
- String[] reducedArguments = new String[] {"--debug", "true", "arg1", "arg2"};
-
- RunOptions options = CliFrontendParser.parseRunCommand(arguments);
- assertEquals(getTestJarPath(), options.getJarFilePath());
- assertArrayEquals(classpath, options.getClasspaths().toArray());
- assertArrayEquals(reducedArguments, options.getProgramArgs());
-
- PackagedProgram prog = frontend.buildProgram(options);
-
- Assert.assertArrayEquals(reducedArguments, prog.getArguments());
- Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ public void testVariantWithExplicitJarAndArgumentsOption() throws Exception {
+ String[] arguments = {
+ "--classpath", "file:///tmp/foo",
+ "--classpath", "file:///tmp/bar",
+ "-j", getTestJarPath(),
+ "-a", "--debug", "true", "arg1", "arg2" };
+ URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
+ String[] reducedArguments = new String[] {"--debug", "true", "arg1", "arg2"};
+
+ RunOptions options = CliFrontendParser.parseRunCommand(arguments);
+ assertEquals(getTestJarPath(), options.getJarFilePath());
+ assertArrayEquals(classpath, options.getClasspaths().toArray());
+ assertArrayEquals(reducedArguments, options.getProgramArgs());
+
+ PackagedProgram prog = frontend.buildProgram(options);
+
+ Assert.assertArrayEquals(reducedArguments, prog.getArguments());
+ Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
}
@Test
- public void testVariantWithExplicitJarAndNoArgumentsOption() {
- try {
- String[] arguments = {
- "--classpath", "file:///tmp/foo",
- "--classpath", "file:///tmp/bar",
- "-j", getTestJarPath(),
- "--debug", "true", "arg1", "arg2" };
- URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
- String[] reducedArguments = new String[] {"--debug", "true", "arg1", "arg2"};
-
- RunOptions options = CliFrontendParser.parseRunCommand(arguments);
- assertEquals(getTestJarPath(), options.getJarFilePath());
- assertArrayEquals(classpath, options.getClasspaths().toArray());
- assertArrayEquals(reducedArguments, options.getProgramArgs());
-
- PackagedProgram prog = frontend.buildProgram(options);
-
- Assert.assertArrayEquals(reducedArguments, prog.getArguments());
- Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ public void testVariantWithExplicitJarAndNoArgumentsOption() throws Exception {
+ String[] arguments = {
+ "--classpath", "file:///tmp/foo",
+ "--classpath", "file:///tmp/bar",
+ "-j", getTestJarPath(),
+ "--debug", "true", "arg1", "arg2" };
+ URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
+ String[] reducedArguments = new String[] {"--debug", "true", "arg1", "arg2"};
+
+ RunOptions options = CliFrontendParser.parseRunCommand(arguments);
+ assertEquals(getTestJarPath(), options.getJarFilePath());
+ assertArrayEquals(classpath, options.getClasspaths().toArray());
+ assertArrayEquals(reducedArguments, options.getProgramArgs());
+
+ PackagedProgram prog = frontend.buildProgram(options);
+
+ Assert.assertArrayEquals(reducedArguments, prog.getArguments());
+ Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
}
@Test
- public void testValidVariantWithNoJarAndNoArgumentsOption() {
- try {
- String[] arguments = {
- "--classpath", "file:///tmp/foo",
- "--classpath", "file:///tmp/bar",
- getTestJarPath(),
- "--debug", "true", "arg1", "arg2" };
- URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
- String[] reducedArguments = {"--debug", "true", "arg1", "arg2"};
-
- RunOptions options = CliFrontendParser.parseRunCommand(arguments);
- assertEquals(getTestJarPath(), options.getJarFilePath());
- assertArrayEquals(classpath, options.getClasspaths().toArray());
- assertArrayEquals(reducedArguments, options.getProgramArgs());
-
- PackagedProgram prog = frontend.buildProgram(options);
-
- Assert.assertArrayEquals(reducedArguments, prog.getArguments());
- Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ public void testValidVariantWithNoJarAndNoArgumentsOption() throws Exception {
+ String[] arguments = {
+ "--classpath", "file:///tmp/foo",
+ "--classpath", "file:///tmp/bar",
+ getTestJarPath(),
+ "--debug", "true", "arg1", "arg2" };
+ URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
+ String[] reducedArguments = {"--debug", "true", "arg1", "arg2"};
+
+ RunOptions options = CliFrontendParser.parseRunCommand(arguments);
+ assertEquals(getTestJarPath(), options.getJarFilePath());
+ assertArrayEquals(classpath, options.getClasspaths().toArray());
+ assertArrayEquals(reducedArguments, options.getProgramArgs());
+
+ PackagedProgram prog = frontend.buildProgram(options);
+
+ Assert.assertArrayEquals(reducedArguments, prog.getArguments());
+ Assert.assertEquals(TEST_JAR_MAIN_CLASS, prog.getMainClassName());
}
@Test(expected = CliArgsException.class)
public void testNoJarNoArgumentsAtAll() throws Exception {
frontend.run(new String[0]);
-
- fail("Should have failed.");
}
@Test
- public void testNonExistingFileWithArguments() {
- try {
- String[] arguments = {
- "--classpath", "file:///tmp/foo",
- "--classpath", "file:///tmp/bar",
- "/some/none/existing/path",
- "--debug", "true", "arg1", "arg2" };
- URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
- String[] reducedArguments = {"--debug", "true", "arg1", "arg2"};
+ public void testNonExistingFileWithArguments() throws Exception {
+ String[] arguments = {
+ "--classpath", "file:///tmp/foo",
+ "--classpath", "file:///tmp/bar",
+ "/some/none/existing/path",
+ "--debug", "true", "arg1", "arg2" };
+ URL[] classpath = new URL[] { new URL("file:///tmp/foo"), new URL("file:///tmp/bar") };
+ String[] reducedArguments = {"--debug", "true", "arg1", "arg2"};
+
+ RunOptions options = CliFrontendParser.parseRunCommand(arguments);
+ assertEquals(arguments[4], options.getJarFilePath());
+ assertArrayEquals(classpath, options.getClasspaths().toArray());
+ assertArrayEquals(reducedArguments, options.getProgramArgs());
- RunOptions options = CliFrontendParser.parseRunCommand(arguments);
- assertEquals(arguments[4], options.getJarFilePath());
- assertArrayEquals(classpath, options.getClasspaths().toArray());
- assertArrayEquals(reducedArguments, options.getProgramArgs());
-
- try {
- frontend.buildProgram(options);
- fail("Should fail with an exception");
- }
- catch (FileNotFoundException e) {
- // that's what we want
- }
+ try {
+ frontend.buildProgram(options);
+ fail("Should fail with an exception");
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+ catch (FileNotFoundException e) {
+ // that's what we want
}
}
@Test
- public void testNonExistingFileWithoutArguments() {
- try {
- String[] arguments = {"/some/none/existing/path"};
+ public void testNonExistingFileWithoutArguments() throws Exception {
+ String[] arguments = {"/some/none/existing/path"};
- RunOptions options = CliFrontendParser.parseRunCommand(arguments);
- assertEquals(arguments[0], options.getJarFilePath());
- assertArrayEquals(new String[0], options.getProgramArgs());
+ RunOptions options = CliFrontendParser.parseRunCommand(arguments);
+ assertEquals(arguments[0], options.getJarFilePath());
+ assertArrayEquals(new String[0], options.getProgramArgs());
- try {
- frontend.buildProgram(options);
- }
- catch (FileNotFoundException e) {
- // that's what we want
- }
+ try {
+ frontend.buildProgram(options);
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+ catch (FileNotFoundException e) {
+ // that's what we want
}
}
@@ -284,7 +239,7 @@ public class CliFrontendPackageProgramTest extends TestLogger {
* </ul>
*/
@Test
- public void testPlanWithExternalClass() throws CompilerException, ProgramInvocationException {
+ public void testPlanWithExternalClass() throws Exception {
final boolean[] callme = { false }; // create a final object reference, to be able to change its val later
try {
@@ -334,9 +289,5 @@ public class CliFrontendPackageProgramTest extends TestLogger {
}
assertTrue("Classloader was not called", callme[0]);
}
- catch (Exception e) {
- e.printStackTrace();
- fail("Program failed with the wrong exception: " + e.getClass().getName());
- }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
----------------------------------------------------------------------
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 fa1c76c..ebb76d8 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
@@ -46,31 +46,32 @@ public class CliFrontendRunTest {
@Test
public void testRun() throws Exception {
+ final Configuration configuration = GlobalConfiguration.loadConfiguration(CliFrontendTestUtils.getConfigDir());
// test without parallelism
{
String[] parameters = {"-v", getTestJarPath()};
- RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(1, true, false);
+ RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(configuration, 1, true, false);
testFrontend.run(parameters);
}
// test configure parallelism
{
String[] parameters = {"-v", "-p", "42", getTestJarPath()};
- RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(42, true, false);
+ RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(configuration, 42, true, false);
testFrontend.run(parameters);
}
// test configure sysout logging
{
String[] parameters = {"-p", "2", "-q", getTestJarPath()};
- RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(2, false, false);
+ RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(configuration, 2, false, false);
testFrontend.run(parameters);
}
// test detached mode
{
String[] parameters = {"-p", "2", "-d", getTestJarPath()};
- RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(2, true, true);
+ RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(configuration, 2, true, true);
testFrontend.run(parameters);
}
@@ -111,10 +112,10 @@ public class CliFrontendRunTest {
public void testUnrecognizedOption() throws Exception {
// test unrecognized option
String[] parameters = {"-v", "-l", "-a", "some", "program", "arguments"};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.run(parameters);
}
@@ -122,10 +123,10 @@ public class CliFrontendRunTest {
public void testInvalidParallelismOption() throws Exception {
// test configure parallelism with non integer value
String[] parameters = {"-v", "-p", "text", getTestJarPath()};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.run(parameters);
}
@@ -133,10 +134,10 @@ public class CliFrontendRunTest {
public void testParallelismWithOverflow() throws Exception {
// test configure parallelism with overflow integer value
String[] parameters = {"-v", "-p", "475871387138", getTestJarPath()};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.run(parameters);
}
@@ -148,11 +149,10 @@ public class CliFrontendRunTest {
private final boolean sysoutLogging;
private final boolean isDetached;
- public RunTestingCliFrontend(int expectedParallelism, boolean logging, boolean isDetached) throws Exception {
+ public RunTestingCliFrontend(Configuration configuration, int expectedParallelism, boolean logging, boolean isDetached) throws Exception {
super(
- GlobalConfiguration.loadConfiguration(CliFrontendTestUtils.getConfigDir()),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
this.expectedParallelism = expectedParallelism;
this.sysoutLogging = logging;
this.isDetached = isDetached;
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendStopTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendStopTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendStopTest.java
index fbbc739..0120cdf 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendStopTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendStopTest.java
@@ -69,10 +69,10 @@ public class CliFrontendStopTest extends TestLogger {
public void testUnrecognizedOption() throws Exception {
// test unrecognized option
String[] parameters = { "-v", "-l" };
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.stop(parameters);
}
@@ -80,10 +80,10 @@ public class CliFrontendStopTest extends TestLogger {
public void testMissingJobId() throws Exception {
// test missing job id
String[] parameters = {};
+ Configuration configuration = new Configuration();
CliFrontend testFrontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
+ configuration,
+ Collections.singletonList(new DefaultCLI(configuration)));
testFrontend.stop(parameters);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java
index 3f05f90..e73b9c9 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java
@@ -46,30 +46,25 @@ public class DefaultCLITest extends TestLogger {
*/
@Test
public void testConfigurationPassing() throws Exception {
- final DefaultCLI defaultCLI = new DefaultCLI();
-
- final String configurationDirectory = temporaryFolder.newFolder().getAbsolutePath();
- final String[] args = {};
-
- CommandLine commandLine = defaultCLI.parseCommandLineOptions(args, false);
+ final Configuration configuration = new Configuration();
final String localhost = "localhost";
final int port = 1234;
- final Configuration configuration = new Configuration();
configuration.setString(JobManagerOptions.ADDRESS, localhost);
configuration.setInteger(JobManagerOptions.PORT, port);
+ final DefaultCLI defaultCLI = new DefaultCLI(configuration);
+
+ final String[] args = {};
+
+ CommandLine commandLine = defaultCLI.parseCommandLineOptions(args, false);
+
final InetSocketAddress expectedAddress = new InetSocketAddress(localhost, port);
- final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(commandLine);
- final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(
- configuration,
- commandLine));
+ final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(commandLine));
Assert.assertEquals(expectedAddress, clusterClient.getJobManagerAddress());
}
@@ -79,15 +74,6 @@ public class DefaultCLITest extends TestLogger {
*/
@Test
public void testManualConfigurationOverride() throws Exception {
- final DefaultCLI defaultCLI = new DefaultCLI();
-
- final String manualHostname = "123.123.123.123";
- final int manualPort = 4321;
- final String configurationDirectory = temporaryFolder.newFolder().getAbsolutePath();
- final String[] args = {"-m", manualHostname + ':' + manualPort};
-
- CommandLine commandLine = defaultCLI.parseCommandLineOptions(args, false);
-
final String localhost = "localhost";
final int port = 1234;
final Configuration configuration = new Configuration();
@@ -95,14 +81,17 @@ public class DefaultCLITest extends TestLogger {
configuration.setString(JobManagerOptions.ADDRESS, localhost);
configuration.setInteger(JobManagerOptions.PORT, port);
- final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(
- configuration,
- configurationDirectory,
- commandLine);
+ final DefaultCLI defaultCLI = new DefaultCLI(configuration);
+
+ final String manualHostname = "123.123.123.123";
+ final int manualPort = 4321;
+ final String[] args = {"-m", manualHostname + ':' + manualPort};
+
+ CommandLine commandLine = defaultCLI.parseCommandLineOptions(args, false);
+
+ final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(commandLine);
- final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(
- configuration,
- commandLine));
+ final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(commandLine));
final InetSocketAddress expectedAddress = new InetSocketAddress(manualHostname, manualPort);
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/Flip6DefaultCLITest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/Flip6DefaultCLITest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/Flip6DefaultCLITest.java
index 6a538d4..c3299bc 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/Flip6DefaultCLITest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/Flip6DefaultCLITest.java
@@ -37,7 +37,7 @@ public class Flip6DefaultCLITest extends TestLogger {
@Test
public void testFlip6Switch() throws CliArgsException {
final String[] args = {"-flip6"};
- final Flip6DefaultCLI flip6DefaultCLI = new Flip6DefaultCLI();
+ final Flip6DefaultCLI flip6DefaultCLI = new Flip6DefaultCLI(new Configuration());
final Options options = new Options();
flip6DefaultCLI.addGeneralOptions(options);
@@ -46,6 +46,6 @@ public class Flip6DefaultCLITest extends TestLogger {
final CommandLine commandLine = CliFrontendParser.parse(options, args, false);
Assert.assertTrue(commandLine.hasOption(Flip6DefaultCLI.FLIP_6.getOpt()));
- Assert.assertTrue(flip6DefaultCLI.isActive(commandLine, new Configuration()));
+ Assert.assertTrue(flip6DefaultCLI.isActive(commandLine));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyCustomCommandLine.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyCustomCommandLine.java b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyCustomCommandLine.java
index f9c2bfa..a36e8e9 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyCustomCommandLine.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyCustomCommandLine.java
@@ -22,7 +22,6 @@ import org.apache.flink.client.cli.CustomCommandLine;
import org.apache.flink.client.deployment.ClusterDescriptor;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Preconditions;
import org.apache.commons.cli.CommandLine;
@@ -43,7 +42,7 @@ public class DummyCustomCommandLine<T extends ClusterClient> implements CustomCo
}
@Override
- public boolean isActive(CommandLine commandLine, Configuration configuration) {
+ public boolean isActive(CommandLine commandLine) {
return true;
}
@@ -63,21 +62,18 @@ public class DummyCustomCommandLine<T extends ClusterClient> implements CustomCo
}
@Override
- public ClusterDescriptor<T> createClusterDescriptor(
- Configuration configuration,
- String configurationDirectory,
- CommandLine commandLine) {
+ public ClusterDescriptor<T> createClusterDescriptor(CommandLine commandLine) {
return new DummyClusterDescriptor<>(clusterClient);
}
@Override
@Nullable
- public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ public String getClusterId(CommandLine commandLine) {
return "dummy";
}
@Override
- public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ public ClusterSpecification getClusterSpecification(CommandLine commandLine) {
return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/cli/util/MockedCliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/util/MockedCliFrontend.java b/flink-clients/src/test/java/org/apache/flink/client/cli/util/MockedCliFrontend.java
index 135e831..00b460f 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/util/MockedCliFrontend.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/util/MockedCliFrontend.java
@@ -19,7 +19,6 @@
package org.apache.flink.client.cli.util;
import org.apache.flink.client.cli.CliFrontend;
-import org.apache.flink.client.cli.CliFrontendTestUtils;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
@@ -35,7 +34,6 @@ public class MockedCliFrontend extends CliFrontend {
public MockedCliFrontend(ClusterClient clusterClient) throws Exception {
super(
new Configuration(),
- Collections.singletonList(new DummyCustomCommandLine<>(clusterClient)),
- CliFrontendTestUtils.getConfigDir());
+ Collections.singletonList(new DummyCustomCommandLine<>(clusterClient)));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-clients/src/test/java/org/apache/flink/client/program/ClusterClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClusterClientTest.java
index ec8091a..7b34d4a 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClusterClientTest.java
@@ -188,6 +188,8 @@ public class ClusterClientTest extends TestLogger {
fail("Dispose operation should have failed.");
} catch (ExecutionException e) {
assertTrue(ExceptionUtils.findThrowable(e, FlinkRuntimeException.class).isPresent());
+ } finally {
+ clusterClient.shutdown();
}
}
@@ -214,6 +216,8 @@ public class ClusterClientTest extends TestLogger {
"instance, which cannot be disposed without the user code class " +
"loader. Please provide the program jar with which you have created " +
"the savepoint via -j <JAR> for disposal.").isPresent());
+ } finally {
+ clusterClient.shutdown();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
index d4d0d05..9f29ce0 100644
--- a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
+++ b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
@@ -256,22 +256,15 @@ object FlinkShell {
val options = CliFrontendParser.parseRunCommand(args.toArray)
val frontend = new CliFrontend(
configuration,
- CliFrontend.loadCustomCommandLines(),
- configurationDirectory)
+ CliFrontend.loadCustomCommandLines(configuration, configurationDirectory))
val config = frontend.getConfiguration
val customCLI = frontend.getActiveCustomCommandLine(options.getCommandLine)
- val clusterDescriptor = customCLI.createClusterDescriptor(
- config,
- frontend.getConfigurationDirectory,
- options.getCommandLine)
+ val clusterDescriptor = customCLI.createClusterDescriptor(options.getCommandLine)
- val clusterSpecification = customCLI.getClusterSpecification(
- config,
- options.getCommandLine)
+ val clusterSpecification = customCLI.getClusterSpecification(options.getCommandLine)
- val cluster = clusterDescriptor.deploySessionCluster(
- clusterSpecification)
+ val cluster = clusterDescriptor.deploySessionCluster(clusterSpecification)
val address = cluster.getJobManagerAddress.getAddress.getHostAddress
val port = cluster.getJobManagerAddress.getPort
@@ -291,19 +284,13 @@ object FlinkShell {
val options = CliFrontendParser.parseRunCommand(args.toArray)
val frontend = new CliFrontend(
configuration,
- CliFrontend.loadCustomCommandLines(),
- configurationDirectory)
+ CliFrontend.loadCustomCommandLines(configuration, configurationDirectory))
val config = frontend.getConfiguration
val customCLI = frontend.getActiveCustomCommandLine(options.getCommandLine)
- val clusterDescriptor = customCLI.createClusterDescriptor(
- configuration,
- configurationDirectory,
- options.getCommandLine)
+ val clusterDescriptor = customCLI.createClusterDescriptor(options.getCommandLine)
- val clusterId = customCLI.getClusterId(
- configuration,
- options.getCommandLine)
+ val clusterId = customCLI.getClusterId(options.getCommandLine)
val cluster = clusterDescriptor.retrieve(clusterId)
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 8716f8a..f347f94 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -577,10 +577,10 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
// load the configuration
LOG.info("testDetachedPerJobYarnClusterInternal: Trying to load configuration file");
- GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
+ Configuration configuration = GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
try {
- File yarnPropertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(GlobalConfiguration.loadConfiguration());
+ File yarnPropertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(configuration.getValue(YarnConfigOptions.PROPERTIES_FILE_LOCATION));
if (yarnPropertiesFile.exists()) {
LOG.info("testDetachedPerJobYarnClusterInternal: Cleaning up temporary Yarn address reference: {}", yarnPropertiesFile.getAbsolutePath());
yarnPropertiesFile.delete();
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index ec8ef50..d5a9883 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -134,10 +135,10 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
// load the configuration
LOG.info("testDetachedPerJobYarnClusterInternal: Trying to load configuration file");
- GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
+ Configuration configuration = GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
try {
- File yarnPropertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(GlobalConfiguration.loadConfiguration());
+ File yarnPropertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(configuration.getString(YarnConfigOptions.PROPERTIES_FILE_LOCATION));
if (yarnPropertiesFile.exists()) {
LOG.info("testDetachedPerJobYarnClusterInternal: Cleaning up temporary Yarn address reference: {}", yarnPropertiesFile.getAbsolutePath());
yarnPropertiesFile.delete();
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
index 5325ae2..ed02892 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
@@ -693,6 +693,8 @@ public abstract class YarnTestBase extends TestLogger {
switch (type) {
case YARN_SESSION:
yCli = new FlinkYarnSessionCli(
+ configuration,
+ configurationDirectory,
"",
"",
false);
@@ -702,8 +704,7 @@ public abstract class YarnTestBase extends TestLogger {
try {
CliFrontend cli = new CliFrontend(
configuration,
- CliFrontend.loadCustomCommandLines(),
- configurationDirectory);
+ CliFrontend.loadCustomCommandLines(configuration, configurationDirectory));
returnValue = cli.parseParameters(args);
} catch (Exception e) {
throw new RuntimeException("Failed to execute the following args with CliFrontend: "
http://git-wip-us.apache.org/repos/asf/flink/blob/30011b9b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index 63421f9..5fb7f90 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
@@ -330,7 +331,7 @@ public class YarnClusterClient extends ClusterClient {
}
try {
- File propertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(flinkConfig);
+ File propertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(flinkConfig.getValue(YarnConfigOptions.PROPERTIES_FILE_LOCATION));
if (propertiesFile.isFile()) {
if (propertiesFile.delete()) {
LOG.info("Deleted Yarn properties file at {}", propertiesFile.getAbsoluteFile().toString());