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/11 23:19:56 UTC
[7/8] flink git commit: [FLINK-8339] [flip6] Let CustomCommandLine
return ClusterDescriptor
[FLINK-8339] [flip6] Let CustomCommandLine return ClusterDescriptor
Instead of directly retrieving or deploying a Flink cluster, the
CustomCommandLine now only returns a ClusterDescriptor which can be used
for these operations. This disentangles the ClusterDescriptor and the
CustomCommandLine a bit better supporting a proper lifecycle management
of the former.
This closes #5225.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e2f1ba92
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e2f1ba92
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e2f1ba92
Branch: refs/heads/master
Commit: e2f1ba92decdb27f3aea4e21a7cad7dcc98cea1a
Parents: aff4376
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Jan 11 16:17:42 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Jan 11 20:40:06 2018 +0100
----------------------------------------------------------------------
.../client/cli/AbstractCustomCommandLine.java | 84 +++
.../apache/flink/client/cli/CliFrontend.java | 323 ++++++------
.../flink/client/cli/CliFrontendParser.java | 37 +-
.../flink/client/cli/CustomCommandLine.java | 69 ++-
.../org/apache/flink/client/cli/DefaultCLI.java | 56 +-
.../flink/client/cli/Flip6DefaultCLI.java | 52 +-
.../flink/client/program/ClusterClient.java | 8 +-
.../client/program/rest/RestClusterClient.java | 27 +-
.../CliFrontendAddressConfigurationTest.java | 84 ---
.../flink/client/cli/CliFrontendCancelTest.java | 17 +-
.../flink/client/cli/CliFrontendInfoTest.java | 10 +-
.../flink/client/cli/CliFrontendListTest.java | 7 +-
.../cli/CliFrontendPackageProgramTest.java | 2 +-
.../flink/client/cli/CliFrontendRunTest.java | 20 +-
.../client/cli/CliFrontendSavepointTest.java | 24 +-
.../flink/client/cli/CliFrontendStopTest.java | 8 +-
.../apache/flink/client/cli/DefaultCLITest.java | 112 ++++
.../client/cli/util/DummyClusterDescriptor.java | 64 +++
.../client/cli/util/DummyCustomCommandLine.java | 22 +-
.../flink/client/program/ClusterClientTest.java | 97 +++-
.../apache/flink/util/ExceptionUtilsTest.java | 2 +-
.../SavepointTriggerResponseBody.java | 12 +
.../org/apache/flink/api/scala/FlinkShell.scala | 34 +-
...CliFrontendYarnAddressConfigurationTest.java | 508 -------------------
.../flink/yarn/FlinkYarnSessionCliTest.java | 217 --------
.../flink/yarn/YarnClusterDescriptorTest.java | 128 -----
.../org/apache/flink/yarn/YarnTestBase.java | 63 +--
.../flink/yarn/cli/FlinkYarnSessionCli.java | 115 +++--
.../flink/yarn/AbstractYarnClusterTest.java | 158 ++++++
.../flink/yarn/FlinkYarnSessionCliTest.java | 294 +++++++++++
.../flink/yarn/YarnClusterDescriptorTest.java | 87 ++++
31 files changed, 1305 insertions(+), 1436 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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
new file mode 100644
index 0000000..70ffc5c
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/AbstractCustomCommandLine.java
@@ -0,0 +1,84 @@
+/*
+ * 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.client.cli;
+
+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.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+
+import java.net.InetSocketAddress;
+
+import static org.apache.flink.client.cli.CliFrontend.setJobManagerAddressInConfig;
+
+/**
+ * Base class for {@link CustomCommandLine} implementations which specify a JobManager address and
+ * a ZooKeeper namespace.
+ *
+ * @param <C> type of the ClusterClient which is returned
+ */
+public abstract class AbstractCustomCommandLine<C extends ClusterClient> implements CustomCommandLine<C> {
+
+ protected final Option zookeeperNamespaceOption = new Option("z", "zookeeperNamespace", true,
+ "Namespace to create the Zookeeper sub-paths for high availability mode");
+
+
+ protected final Option addressOption = new Option("m", "jobmanager", true,
+ "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.");
+
+ @Override
+ public void addRunOptions(Options baseOptions) {
+ // nothing to add here
+ }
+
+ @Override
+ public void addGeneralOptions(Options baseOptions) {
+ baseOptions.addOption(addressOption);
+ baseOptions.addOption(zookeeperNamespaceOption);
+ }
+
+ /**
+ * 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) {
+ final Configuration resultingConfiguration = new Configuration(configuration);
+
+ if (commandLine.hasOption(addressOption.getOpt())) {
+ String addressWithPort = commandLine.getOptionValue(addressOption.getOpt());
+ InetSocketAddress jobManagerAddress = ClientUtils.parseHostPortAddress(addressWithPort);
+ setJobManagerAddressInConfig(resultingConfiguration, jobManagerAddress);
+ }
+
+ if (commandLine.hasOption(zookeeperNamespaceOption.getOpt())) {
+ String zkNamespace = commandLine.getOptionValue(zookeeperNamespaceOption.getOpt());
+ resultingConfiguration.setString(HighAvailabilityOptions.HA_CLUSTER_ID, zkNamespace);
+ }
+
+ return resultingConfiguration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 e539240..4b5040e 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
@@ -24,6 +24,8 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobSubmissionResult;
import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+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.client.program.PackagedProgram;
import org.apache.flink.client.program.ProgramInvocationException;
@@ -32,7 +34,6 @@ import org.apache.flink.client.program.ProgramParametrizationException;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
-import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.optimizer.DataStatistics;
@@ -178,7 +179,7 @@ public class CliFrontend {
*
* @param args Command line arguments for the run action.
*/
- protected int run(String[] args) throws Exception {
+ protected void run(String[] args) throws Exception {
LOG.info("Running 'run' command.");
final Options commandOptions = CliFrontendParser.getRunCommandOptions();
@@ -192,7 +193,7 @@ public class CliFrontend {
// evaluate help flag
if (runOptions.isPrintHelp()) {
CliFrontendParser.printHelpForRun(customCommandLines);
- return 0;
+ return;
}
if (runOptions.getJarFilePath() == null) {
@@ -208,41 +209,59 @@ public class CliFrontend {
throw new CliArgsException("Could not build the program from JAR file.", e);
}
- ClusterClient client = null;
+ final CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
+
+ final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
try {
- final CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
- client = createClient(customCommandLine, commandLine, program);
-
- client.setPrintStatusDuringExecution(runOptions.getStdoutLogging());
- client.setDetached(runOptions.getDetachedMode());
- LOG.debug("Client slots is set to {}", client.getMaxSlots());
-
- LOG.debug(runOptions.getSavepointRestoreSettings().toString());
-
- int userParallelism = runOptions.getParallelism();
- LOG.debug("User parallelism is set to {}", userParallelism);
- if (client.getMaxSlots() != -1 && userParallelism == -1) {
- logAndSysout("Using the parallelism provided by the remote cluster ("
- + client.getMaxSlots() + "). "
- + "To use another parallelism, set it at the ./bin/flink client.");
- userParallelism = client.getMaxSlots();
- } else if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
- userParallelism = defaultParallelism;
+ final String clusterId = customCommandLine.getClusterId(configuration, commandLine);
+
+ final ClusterClient client;
+
+ if (clusterId != null) {
+ client = clusterDescriptor.retrieve(clusterId);
+ } else {
+ final ClusterSpecification clusterSpecification = customCommandLine.getClusterSpecification(configuration, commandLine);
+ client = clusterDescriptor.deploySessionCluster(clusterSpecification);
}
- return executeProgram(program, client, userParallelism);
- }
- finally {
- program.deleteExtractedLibraries();
+ try {
+ client.setPrintStatusDuringExecution(runOptions.getStdoutLogging());
+ client.setDetached(runOptions.getDetachedMode());
+ LOG.debug("Client slots is set to {}", client.getMaxSlots());
+
+ LOG.debug(runOptions.getSavepointRestoreSettings().toString());
+
+ int userParallelism = runOptions.getParallelism();
+ LOG.debug("User parallelism is set to {}", userParallelism);
+ if (client.getMaxSlots() != -1 && userParallelism == -1) {
+ logAndSysout("Using the parallelism provided by the remote cluster ("
+ + client.getMaxSlots() + "). "
+ + "To use another parallelism, set it at the ./bin/flink client.");
+ userParallelism = client.getMaxSlots();
+ } else if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
+ userParallelism = defaultParallelism;
+ }
- if (client != null) {
+ executeProgram(program, client, userParallelism);
+ } finally {
try {
client.shutdown();
} catch (Exception e) {
LOG.info("Could not properly shut down the client.", e);
}
}
+ } finally {
+ program.deleteExtractedLibraries();
+
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
}
}
@@ -251,7 +270,7 @@ public class CliFrontend {
*
* @param args Command line arguments for the info action.
*/
- protected int info(String[] args) throws CliArgsException, FileNotFoundException, ProgramInvocationException {
+ protected void info(String[] args) throws CliArgsException, FileNotFoundException, ProgramInvocationException {
LOG.info("Running 'info' command.");
final Options commandOptions = CliFrontendParser.getInfoCommandOptions();
@@ -263,11 +282,11 @@ public class CliFrontend {
// evaluate help flag
if (infoOptions.isPrintHelp()) {
CliFrontendParser.printHelpForInfo();
- return 0;
+ return;
}
if (infoOptions.getJarFilePath() == null) {
- return handleArgException(new CliArgsException("The program JAR file was not specified."));
+ throw new CliArgsException("The program JAR file was not specified.");
}
// -------- build the packaged program -------------
@@ -311,7 +330,6 @@ public class CliFrontend {
System.out.println();
System.out.println("No description provided.");
}
- return 0;
}
finally {
program.deleteExtractedLibraries();
@@ -323,7 +341,7 @@ public class CliFrontend {
*
* @param args Command line arguments for the list action.
*/
- protected int list(String[] args) throws Exception {
+ protected void list(String[] args) throws Exception {
LOG.info("Running 'list' command.");
final Options commandOptions = CliFrontendParser.getListCommandOptions();
@@ -337,7 +355,7 @@ public class CliFrontend {
// evaluate help flag
if (listOptions.isPrintHelp()) {
CliFrontendParser.printHelpForList(customCommandLines);
- return 0;
+ return;
}
boolean running = listOptions.getRunning();
@@ -350,7 +368,21 @@ public class CliFrontend {
}
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterClient client = activeCommandLine.retrieveCluster(commandLine, configuration, configurationDirectory);
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
+
+ final String clusterId = activeCommandLine.getClusterId(
+ configuration,
+ commandLine);
+
+ if (clusterId == null) {
+ throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
+ "you would like to connect.");
+ }
+
+ final ClusterClient client = clusterDescriptor.retrieve(clusterId);
try {
Collection<JobStatusMessage> jobDetails;
@@ -414,14 +446,18 @@ public class CliFrontend {
System.out.println("--------------------------------------------------------------");
}
}
-
- return 0;
} finally {
try {
client.shutdown();
} catch (Exception e) {
LOG.info("Could not properly shut down the client.", e);
}
+
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
}
}
@@ -430,7 +466,7 @@ public class CliFrontend {
*
* @param args Command line arguments for the stop action.
*/
- protected int stop(String[] args) throws Exception {
+ protected void stop(String[] args) throws Exception {
LOG.info("Running 'stop' command.");
final Options commandOptions = CliFrontendParser.getStopCommandOptions();
@@ -444,7 +480,7 @@ public class CliFrontend {
// evaluate help flag
if (stopOptions.isPrintHelp()) {
CliFrontendParser.printHelpForStop(customCommandLines);
- return 0;
+ return;
}
String[] stopArgs = stopOptions.getArgs();
@@ -460,20 +496,36 @@ public class CliFrontend {
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterClient client = activeCommandLine.retrieveCluster(commandLine, configuration, configurationDirectory);
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
+
+ final String clusterId = activeCommandLine.getClusterId(configuration, commandLine);
+
+ if (clusterId == null) {
+ throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
+ "you would like to connect.");
+ }
+
+ final ClusterClient client = clusterDescriptor.retrieve(clusterId);
try {
logAndSysout("Stopping job " + jobId + '.');
client.stop(jobId);
logAndSysout("Stopped job " + jobId + '.');
-
- return 0;
} finally {
try {
client.shutdown();
} catch (Exception e) {
LOG.info("Could not properly shut down the client.", e);
}
+
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
}
}
@@ -482,7 +534,7 @@ public class CliFrontend {
*
* @param args Command line arguments for the cancel action.
*/
- protected int cancel(String[] args) throws Exception {
+ protected void cancel(String[] args) throws Exception {
LOG.info("Running 'cancel' command.");
final Options commandOptions = CliFrontendParser.getCancelCommandOptions();
@@ -496,7 +548,7 @@ public class CliFrontend {
// evaluate help flag
if (cancelOptions.isPrintHelp()) {
CliFrontendParser.printHelpForCancel(customCommandLines);
- return 0;
+ return;
}
String[] cleanedArgs = cancelOptions.getArgs();
@@ -526,7 +578,20 @@ public class CliFrontend {
}
final CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(commandLine);
- final ClusterClient client = activeCommandLine.retrieveCluster(commandLine, configuration, configurationDirectory);
+
+ final ClusterDescriptor<?> clusterDescriptor = activeCommandLine.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
+
+ final String clusterId = activeCommandLine.getClusterId(configuration, commandLine);
+
+ if (clusterId == null) {
+ throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
+ "you would like to connect.");
+ }
+
+ final ClusterClient client = clusterDescriptor.retrieve(clusterId);
try {
if (withSavepoint) {
@@ -542,23 +607,27 @@ public class CliFrontend {
client.cancel(jobId);
logAndSysout("Cancelled job " + jobId + '.');
}
-
- return 0;
} finally {
try {
client.shutdown();
} catch (Exception e) {
LOG.info("Could not properly shut down the client.", e);
}
+
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
}
}
/**
* Executes the SAVEPOINT action.
*
- * @param args Command line arguments for the cancel action.
+ * @param args Command line arguments for the savepoint action.
*/
- protected int savepoint(String[] args) throws Exception {
+ protected void savepoint(String[] args) throws Exception {
LOG.info("Running 'savepoint' command.");
final Options commandOptions = CliFrontendParser.getSavepointCommandOptions();
@@ -572,20 +641,31 @@ public class CliFrontend {
// evaluate help flag
if (savepointOptions.isPrintHelp()) {
CliFrontendParser.printHelpForSavepoint(customCommandLines);
- return 0;
+ return;
}
CustomCommandLine<?> customCommandLine = getActiveCustomCommandLine(commandLine);
- ClusterClient clusterClient = customCommandLine.retrieveCluster(
- commandLine,
+ final ClusterDescriptor<?> clusterDescriptor = customCommandLine.createClusterDescriptor(
configuration,
- configurationDirectory);
+ configurationDirectory,
+ commandLine);
+
+ final String clusterId = customCommandLine.getClusterId(
+ configuration,
+ commandLine);
+
+ if (clusterId == null) {
+ throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
+ "you would like to connect.");
+ }
+
+ final ClusterClient clusterClient = clusterDescriptor.retrieve(clusterId);
try {
if (savepointOptions.isDispose()) {
// Discard
- return disposeSavepoint(clusterClient, savepointOptions.getSavepointPath());
+ disposeSavepoint(clusterClient, savepointOptions.getSavepointPath());
} else {
// Trigger
String[] cleanedArgs = savepointOptions.getArgs();
@@ -610,7 +690,7 @@ public class CliFrontend {
logAndSysout("Provided more arguments than required. Ignoring not needed arguments.");
}
- return triggerSavepoint(clusterClient, jobId, savepointDirectory);
+ triggerSavepoint(clusterClient, jobId, savepointDirectory);
}
} finally {
try {
@@ -618,6 +698,12 @@ public class CliFrontend {
} catch (Exception e) {
LOG.info("Could not shutdown the cluster client.", e);
}
+
+ try {
+ clusterDescriptor.close();
+ } catch (Exception e) {
+ LOG.info("Could not properly close the cluster descriptor.", e);
+ }
}
}
@@ -625,78 +711,62 @@ public class CliFrontend {
* Sends a {@link org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint}
* message to the job manager.
*/
- private int triggerSavepoint(ClusterClient clusterClient, JobID jobId, String savepointDirectory) throws Exception {
- logAndSysout("Triggering savepoint for job " + jobId + ".");
+ private String triggerSavepoint(ClusterClient clusterClient, JobID jobId, String savepointDirectory) throws FlinkException {
+ logAndSysout("Triggering savepoint for job " + jobId + '.');
CompletableFuture<String> savepointPathFuture = clusterClient.triggerSavepoint(jobId, savepointDirectory);
- String savepointPath;
+ logAndSysout("Waiting for response...");
+
+ final String savepointPath;
+
try {
- logAndSysout("Waiting for response...");
savepointPath = savepointPathFuture.get();
}
- catch (ExecutionException ee) {
- Throwable cause = ExceptionUtils.stripExecutionException(ee);
+ catch (Exception e) {
+ Throwable cause = ExceptionUtils.stripExecutionException(e);
throw new FlinkException("Triggering a savepoint for the job " + jobId + " failed.", cause);
}
logAndSysout("Savepoint completed. Path: " + savepointPath);
logAndSysout("You can resume your program from this savepoint with the run command.");
- return 0;
+ return savepointPath;
}
/**
* Sends a {@link JobManagerMessages.DisposeSavepoint} message to the job manager.
*/
- private int disposeSavepoint(ClusterClient clusterClient, String savepointPath) {
+ private void disposeSavepoint(ClusterClient clusterClient, String savepointPath) throws FlinkException {
Preconditions.checkNotNull(savepointPath, "Missing required argument: savepoint path. " +
"Usage: bin/flink savepoint -d <savepoint-path>");
logAndSysout("Disposing savepoint '" + savepointPath + "'.");
- final CompletableFuture<Acknowledge> disposeFuture;
- try {
- disposeFuture = clusterClient.disposeSavepoint(savepointPath, FutureUtils.toTime(clientTimeout));
- } catch (Exception e) {
- return handleError(new FlinkException("Could not dispose savepoint", e));
- }
+ final CompletableFuture<Acknowledge> disposeFuture = clusterClient.disposeSavepoint(savepointPath, FutureUtils.toTime(clientTimeout));
logAndSysout("Waiting for response...");
try {
disposeFuture.get(clientTimeout.toMillis(), TimeUnit.MILLISECONDS);
} catch (Exception e) {
- return handleError(e);
+ throw new FlinkException("Disposing the savepoint '" + savepointPath + "' failed.", e);
}
logAndSysout("Savepoint '" + savepointPath + "' disposed.");
- return 0;
}
// --------------------------------------------------------------------------------------------
// Interaction with programs and JobManager
// --------------------------------------------------------------------------------------------
- protected int executeProgram(PackagedProgram program, ClusterClient client, int parallelism) {
+ protected void executeProgram(PackagedProgram program, ClusterClient client, int parallelism) throws ProgramMissingJobException, ProgramInvocationException {
logAndSysout("Starting execution of program");
- JobSubmissionResult result;
- try {
- result = client.run(program, parallelism);
- } catch (ProgramParametrizationException e) {
- return handleParametrizationException(e);
- } catch (ProgramMissingJobException e) {
- return handleMissingJobException();
- } catch (ProgramInvocationException e) {
- return handleError(e);
- } finally {
- program.deleteExtractedLibraries();
- }
+ final JobSubmissionResult result = client.run(program, parallelism);
if (null == result) {
- logAndSysout("No JobSubmissionResult returned, please make sure you called " +
+ throw new ProgramMissingJobException("No JobSubmissionResult returned, please make sure you called " +
"ExecutionEnvironment.execute()");
- return 1;
}
if (result.isJobExecutionResult()) {
@@ -712,8 +782,6 @@ public class CliFrontend {
} else {
logAndSysout("Job has been submitted with JobID " + result.getJobID());
}
-
- return 0;
}
/**
@@ -755,67 +823,6 @@ public class CliFrontend {
return program;
}
- /**
- * Updates the associated configuration with the given command line options.
- *
- * @param options Command line options
- */
- protected ClusterClient retrieveClient(CommandLineOptions options) {
- CustomCommandLine customCLI = getActiveCustomCommandLine(options.getCommandLine());
- try {
- ClusterClient client = customCLI.retrieveCluster(options.getCommandLine(), configuration, configurationDirectory);
- logAndSysout("Using address " + client.getJobManagerAddress() + " to connect to JobManager.");
- return client;
- } catch (Exception e) {
- LOG.error("Couldn't retrieve {} cluster.", customCLI.getId(), e);
- throw new IllegalConfigurationException("Couldn't retrieve client for cluster", e);
- }
- }
-
- /**
- * Creates a {@link ClusterClient} object from the given command line options and other parameters.
- * @param customCommandLine custom command line to use to retrieve the client
- * @param commandLine command line to use
- * @param program The program for which to create the client.
- * @throws Exception
- */
- protected ClusterClient createClient(
- CustomCommandLine<?> customCommandLine,
- CommandLine commandLine,
- PackagedProgram program) throws Exception {
-
- ClusterClient client;
- try {
- client = customCommandLine.retrieveCluster(commandLine, configuration, configurationDirectory);
- logAndSysout("Cluster configuration: " + client.getClusterIdentifier());
- } catch (UnsupportedOperationException e) {
- try {
- String applicationName = "Flink Application: " + program.getMainClassName();
- client = customCommandLine.createCluster(
- applicationName,
- commandLine,
- configuration,
- configurationDirectory,
- program.getAllLibraries());
- logAndSysout("Cluster started: " + client.getClusterIdentifier());
- } catch (UnsupportedOperationException e2) {
- throw new IllegalConfigurationException(
- "The JobManager address is neither provided at the command-line, " +
- "nor configured in flink-conf.yaml.");
- }
- }
-
- // Avoid resolving the JobManager Gateway here to prevent blocking until we invoke the user's program.
- final InetSocketAddress jobManagerAddress = client.getJobManagerAddress();
- logAndSysout("Using address " + jobManagerAddress.getHostString() + ":" + jobManagerAddress.getPort() + " to connect to JobManager.");
- try {
- logAndSysout("JobManager web interface address " + client.getWebInterfaceURL());
- } catch (UnsupportedOperationException uoe) {
- logAndSysout("JobManager web interface not active.");
- }
- return client;
- }
-
// --------------------------------------------------------------------------------------------
// Logging and Exception Handling
// --------------------------------------------------------------------------------------------
@@ -826,7 +833,7 @@ public class CliFrontend {
* @param e The exception to display.
* @return The return code for the process.
*/
- private static int handleArgException(Exception e) {
+ private static int handleArgException(CliArgsException e) {
LOG.error("Invalid command line arguments. " + (e.getMessage() == null ? "" : e.getMessage()));
System.out.println(e.getMessage());
@@ -935,17 +942,23 @@ public class CliFrontend {
// do action
switch (action) {
case ACTION_RUN:
- return run(params);
+ run(params);
+ return 0;
case ACTION_LIST:
- return list(params);
+ list(params);
+ return 0;
case ACTION_INFO:
- return info(params);
+ info(params);
+ return 0;
case ACTION_CANCEL:
- return cancel(params);
+ cancel(params);
+ return 0;
case ACTION_STOP:
- return stop(params);
+ stop(params);
+ return 0;
case ACTION_SAVEPOINT:
- return savepoint(params);
+ savepoint(params);
+ return 0;
case "-h":
case "--help":
CliFrontendParser.printHelp(customCommandLines);
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index c6e705d..475d854 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -153,7 +153,6 @@ public class CliFrontendParser {
options.addOption(ARGS_OPTION);
options.addOption(LOGGING_OPTION);
options.addOption(DETACHED_OPTION);
- options.addOption(ZOOKEEPER_NAMESPACE_OPTION);
return options;
}
@@ -163,7 +162,6 @@ public class CliFrontendParser {
options.addOption(PARALLELISM_OPTION);
options.addOption(LOGGING_OPTION);
options.addOption(DETACHED_OPTION);
- options.addOption(ZOOKEEPER_NAMESPACE_OPTION);
return options;
}
@@ -171,38 +169,27 @@ public class CliFrontendParser {
Options options = buildGeneralOptions(new Options());
options = getProgramSpecificOptions(options);
options.addOption(SAVEPOINT_PATH_OPTION);
- options.addOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION);
-
- return getJobManagerAddressOption(options);
- }
-
- private static Options getJobManagerAddressOption(Options options) {
- options.addOption(ADDRESS_OPTION);
- return options;
+ return options.addOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION);
}
static Options getInfoCommandOptions() {
Options options = buildGeneralOptions(new Options());
- options = getProgramSpecificOptions(options);
- return getJobManagerAddressOption(options);
+ return getProgramSpecificOptions(options);
}
static Options getListCommandOptions() {
Options options = buildGeneralOptions(new Options());
options.addOption(RUNNING_OPTION);
- options.addOption(SCHEDULED_OPTION);
- return getJobManagerAddressOption(options);
+ return options.addOption(SCHEDULED_OPTION);
}
static Options getCancelCommandOptions() {
Options options = buildGeneralOptions(new Options());
- options.addOption(CANCEL_WITH_SAVEPOINT_OPTION);
- return getJobManagerAddressOption(options);
+ return options.addOption(CANCEL_WITH_SAVEPOINT_OPTION);
}
static Options getStopCommandOptions() {
- Options options = buildGeneralOptions(new Options());
- return getJobManagerAddressOption(options);
+ return buildGeneralOptions(new Options());
}
static Options getSavepointCommandOptions() {
@@ -218,9 +205,7 @@ public class CliFrontendParser {
private static Options getRunOptionsWithoutDeprecatedOptions(Options options) {
Options o = getProgramSpecificOptionsWithoutDeprecatedOptions(options);
o.addOption(SAVEPOINT_PATH_OPTION);
- o.addOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION);
-
- return getJobManagerAddressOption(o);
+ return o.addOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION);
}
private static Options getInfoOptionsWithoutDeprecatedOptions(Options options) {
@@ -231,24 +216,18 @@ public class CliFrontendParser {
private static Options getListOptionsWithoutDeprecatedOptions(Options options) {
options.addOption(RUNNING_OPTION);
- options.addOption(SCHEDULED_OPTION);
- options = getJobManagerAddressOption(options);
- return options;
+ return options.addOption(SCHEDULED_OPTION);
}
private static Options getCancelOptionsWithoutDeprecatedOptions(Options options) {
- options.addOption(CANCEL_WITH_SAVEPOINT_OPTION);
- options = getJobManagerAddressOption(options);
- return options;
+ return options.addOption(CANCEL_WITH_SAVEPOINT_OPTION);
}
private static Options getStopOptionsWithoutDeprecatedOptions(Options options) {
- options = getJobManagerAddressOption(options);
return options;
}
private static Options getSavepointOptionsWithoutDeprecatedOptions(Options options) {
- options = getJobManagerAddressOption(options);
options.addOption(SAVEPOINT_DISPOSE_OPTION);
options.addOption(JAR_OPTION);
return options;
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 517fc71..cfa0827 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
@@ -18,14 +18,15 @@
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.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
-import java.net.URL;
-import java.util.List;
+import javax.annotation.Nullable;
/**
* Custom command-line interface to load hooks for the command-line interface.
@@ -59,31 +60,49 @@ public interface CustomCommandLine<ClusterType extends ClusterClient> {
void addGeneralOptions(Options baseOptions);
/**
- * Retrieves a client for a running cluster.
- * @param commandLine The command-line parameters from the CliFrontend
- * @param config The Flink config
- * @param configurationDirectory Directory for configuration files
- * @return Client if a cluster could be retrieved
- * @throws UnsupportedOperationException if the operation is not supported
+ * 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
*/
- ClusterType retrieveCluster(
- CommandLine commandLine,
- Configuration config,
- String configurationDirectory) throws UnsupportedOperationException;
+ ClusterDescriptor<ClusterType> createClusterDescriptor(
+ Configuration configuration,
+ String configurationDirectory,
+ CommandLine commandLine);
/**
- * Creates the client for the cluster.
- * @param applicationName The application name to use
- * @param commandLine The command-line options parsed by the CliFrontend
- * @param config The Flink config to use
- * @param configurationDirectory Directory for configuration files
- *@param userJarFiles User jar files to include in the classpath of the cluster. @return The client to communicate with the cluster which the CustomCommandLine brought up.
- * @throws Exception if the cluster could not be created
+ * Returns the cluster id if a cluster id was specified on the command line, otherwise it
+ * returns null.
+ *
+ * <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
*/
- ClusterType createCluster(
- String applicationName,
- CommandLine commandLine,
- Configuration config,
- String configurationDirectory,
- List<URL> userJarFiles) throws Exception;
+ @Nullable
+ String getClusterId(Configuration configuration, 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
+ */
+ ClusterSpecification getClusterSpecification(
+ Configuration configuration,
+ CommandLine commandLine);
+
+ default CommandLine parseCommandLineOptions(String[] args, boolean stopAtNonOptions) throws CliArgsException {
+ final Options options = new Options();
+ addGeneralOptions(options);
+ addRunOptions(options);
+ return CliFrontendParser.parse(options, args, stopAtNonOptions);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 1d25752..d34e307 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
@@ -18,26 +18,20 @@
package org.apache.flink.client.cli;
-import org.apache.flink.client.ClientUtils;
+import org.apache.flink.client.deployment.ClusterDescriptor;
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.configuration.HighAvailabilityOptions;
import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Options;
-import java.net.InetSocketAddress;
-import java.net.URL;
-import java.util.List;
-
-import static org.apache.flink.client.cli.CliFrontend.setJobManagerAddressInConfig;
+import javax.annotation.Nullable;
/**
* The default CLI which is used for interaction with standalone clusters.
*/
-public class DefaultCLI implements CustomCommandLine<StandaloneClusterClient> {
+public class DefaultCLI extends AbstractCustomCommandLine<StandaloneClusterClient> {
@Override
public boolean isActive(CommandLine commandLine, Configuration configuration) {
@@ -51,45 +45,23 @@ public class DefaultCLI implements CustomCommandLine<StandaloneClusterClient> {
}
@Override
- public void addRunOptions(Options baseOptions) {
- }
+ public ClusterDescriptor<StandaloneClusterClient> createClusterDescriptor(
+ Configuration configuration,
+ String configurationDirectory,
+ CommandLine commandLine) {
+ final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(configuration, commandLine);
- @Override
- public void addGeneralOptions(Options baseOptions) {
+ return new StandaloneClusterDescriptor(effectiveConfiguration);
}
@Override
- public StandaloneClusterClient retrieveCluster(
- CommandLine commandLine,
- Configuration config,
- String configurationDirectory) {
-
- if (commandLine.hasOption(CliFrontendParser.ADDRESS_OPTION.getOpt())) {
- String addressWithPort = commandLine.getOptionValue(CliFrontendParser.ADDRESS_OPTION.getOpt());
- InetSocketAddress jobManagerAddress = ClientUtils.parseHostPortAddress(addressWithPort);
- setJobManagerAddressInConfig(config, jobManagerAddress);
- }
-
- if (commandLine.hasOption(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt())) {
- String zkNamespace = commandLine.getOptionValue(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt());
- config.setString(HighAvailabilityOptions.HA_CLUSTER_ID, zkNamespace);
- }
-
- StandaloneClusterDescriptor descriptor = new StandaloneClusterDescriptor(config);
- return descriptor.retrieve(null);
+ @Nullable
+ public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ return "standalone";
}
@Override
- public StandaloneClusterClient createCluster(
- String applicationName,
- CommandLine commandLine,
- Configuration config,
- String configurationDirectory,
- List<URL> userJarFiles) throws UnsupportedOperationException {
-
- StandaloneClusterDescriptor descriptor = new StandaloneClusterDescriptor(config);
- ClusterSpecification clusterSpecification = ClusterSpecification.fromConfiguration(config);
-
- return descriptor.deploySessionCluster(clusterSpecification);
+ public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 413b465..0e80f44 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
@@ -18,27 +18,22 @@
package org.apache.flink.client.cli;
-import org.apache.flink.client.ClientUtils;
+import org.apache.flink.client.deployment.ClusterDescriptor;
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.configuration.HighAvailabilityOptions;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
-import java.net.InetSocketAddress;
-import java.net.URL;
-import java.util.List;
-
-import static org.apache.flink.client.cli.CliFrontend.setJobManagerAddressInConfig;
+import javax.annotation.Nullable;
/**
* The default CLI which is used for interaction with standalone clusters.
*/
-public class Flip6DefaultCLI implements CustomCommandLine<RestClusterClient> {
+public class Flip6DefaultCLI extends AbstractCustomCommandLine<RestClusterClient> {
public static final Option FLIP_6 = new Option("flip6", "Switches the client to Flip-6 mode.");
@@ -57,42 +52,29 @@ public class Flip6DefaultCLI implements CustomCommandLine<RestClusterClient> {
}
@Override
- public void addRunOptions(Options baseOptions) {
- }
-
- @Override
public void addGeneralOptions(Options baseOptions) {
+ super.addGeneralOptions(baseOptions);
baseOptions.addOption(FLIP_6);
}
@Override
- public RestClusterClient retrieveCluster(CommandLine commandLine, Configuration config, String configurationDirectory) {
- if (commandLine.hasOption(CliFrontendParser.ADDRESS_OPTION.getOpt())) {
- String addressWithPort = commandLine.getOptionValue(CliFrontendParser.ADDRESS_OPTION.getOpt());
- InetSocketAddress jobManagerAddress = ClientUtils.parseHostPortAddress(addressWithPort);
- setJobManagerAddressInConfig(config, jobManagerAddress);
- }
-
- if (commandLine.hasOption(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt())) {
- String zkNamespace = commandLine.getOptionValue(CliFrontendParser.ZOOKEEPER_NAMESPACE_OPTION.getOpt());
- config.setString(HighAvailabilityOptions.HA_CLUSTER_ID, zkNamespace);
- }
+ public ClusterDescriptor<RestClusterClient> createClusterDescriptor(
+ Configuration configuration,
+ String configurationDirectory,
+ CommandLine commandLine) {
+ final Configuration effectiveConfiguration = applyCommandLineOptionsToConfiguration(configuration, commandLine);
- Flip6StandaloneClusterDescriptor descriptor = new Flip6StandaloneClusterDescriptor(config);
- return descriptor.retrieve(null);
+ return new Flip6StandaloneClusterDescriptor(effectiveConfiguration);
}
@Override
- public RestClusterClient createCluster(
- String applicationName,
- CommandLine commandLine,
- Configuration config,
- String configurationDirectory,
- List<URL> userJarFiles) throws UnsupportedOperationException {
-
- Flip6StandaloneClusterDescriptor descriptor = new Flip6StandaloneClusterDescriptor(config);
- ClusterSpecification clusterSpecification = ClusterSpecification.fromConfiguration(config);
+ @Nullable
+ public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ return "flip6Standalone";
+ }
- return descriptor.deploySessionCluster(clusterSpecification);
+ @Override
+ public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index b91927a..5f0c25f 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -670,9 +670,9 @@ public abstract class ClusterClient {
* @param jobId job id
* @param savepointDirectory directory the savepoint should be written to
* @return path future where the savepoint is located
- * @throws Exception if no connection to the cluster could be established
+ * @throws FlinkException if no connection to the cluster could be established
*/
- public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception {
+ public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws FlinkException {
final ActorGateway jobManager = getJobManagerGateway();
Future<Object> response = jobManager.ask(new JobManagerMessages.TriggerSavepoint(jobId, Option.apply(savepointDirectory)),
@@ -693,7 +693,7 @@ public abstract class ClusterClient {
});
}
- public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath, Time timeout) throws Exception {
+ public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath, Time timeout) throws FlinkException {
final ActorGateway jobManager = getJobManagerGateway();
Object msg = new JobManagerMessages.DisposeSavepoint(savepointPath);
@@ -887,7 +887,7 @@ public abstract class ClusterClient {
* @return ActorGateway of the current job manager leader
* @throws Exception
*/
- public ActorGateway getJobManagerGateway() throws Exception {
+ public ActorGateway getJobManagerGateway() throws FlinkException {
log.debug("Looking up JobManager");
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index eedcee1..a7d9d02 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -49,9 +49,11 @@ import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerHea
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.FlinkException;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.ArrayList;
@@ -182,21 +184,28 @@ public class RestClusterClient extends ClusterClient {
}
@Override
- public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws Exception {
+ public CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory) throws FlinkException {
SavepointTriggerHeaders headers = SavepointTriggerHeaders.getInstance();
SavepointMessageParameters params = headers.getUnresolvedMessageParameters();
params.jobID.resolve(jobId);
+
if (savepointDirectory != null) {
params.targetDirectory.resolve(Collections.singletonList(savepointDirectory));
}
- CompletableFuture<SavepointTriggerResponseBody> responseFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- headers,
- params
- );
- return responseFuture
- .thenApply(response -> response.location);
+
+ final CompletableFuture<SavepointTriggerResponseBody> responseFuture;
+
+ try {
+ responseFuture = restClient.sendRequest(
+ restClusterClientConfiguration.getRestServerAddress(),
+ restClusterClientConfiguration.getRestServerPort(),
+ headers,
+ params);
+ } catch (IOException e) {
+ throw new FlinkException("Could not send trigger savepoint request to Flink cluster.", e);
+ }
+
+ return responseFuture.thenApply(SavepointTriggerResponseBody::getLocation);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendAddressConfigurationTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendAddressConfigurationTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendAddressConfigurationTest.java
deleted file mode 100644
index 97fb28d..0000000
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendAddressConfigurationTest.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.client.cli;
-
-import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.GlobalConfiguration;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-
-import java.net.InetSocketAddress;
-import java.util.Collections;
-
-import static org.apache.flink.client.cli.CliFrontendTestUtils.checkJobManagerAddress;
-
-/**
- * Tests that verify that the CLI client picks up the correct address for the JobManager
- * from configuration and configs.
- */
-public class CliFrontendAddressConfigurationTest extends TestLogger {
-
- @Rule
- public TemporaryFolder folder = new TemporaryFolder();
-
- @BeforeClass
- public static void init() {
- CliFrontendTestUtils.pipeSystemOutToNull();
- }
-
- @Test
- public void testValidConfig() throws Exception {
- CliFrontend frontend = new CliFrontend(
- GlobalConfiguration.loadConfiguration(CliFrontendTestUtils.getConfigDir()),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
-
- RunOptions options = CliFrontendParser.parseRunCommand(new String[] {});
-
- ClusterClient clusterClient = frontend.retrieveClient(options);
-
- checkJobManagerAddress(
- clusterClient.getFlinkConfiguration(),
- CliFrontendTestUtils.TEST_JOB_MANAGER_ADDRESS,
- CliFrontendTestUtils.TEST_JOB_MANAGER_PORT);
- }
-
- @Test
- public void testManualOptionsOverridesConfig() throws Exception {
- CliFrontend frontend = new CliFrontend(
- new Configuration(),
- Collections.singletonList(new DefaultCLI()),
- CliFrontendTestUtils.getConfigDir());
-
- RunOptions options = CliFrontendParser.parseRunCommand(new String[] {"-m", "203.0.113.22:7788"});
-
- ClusterClient client = frontend.retrieveClient(options);
- Configuration config = client.getFlinkConfiguration();
-
- InetSocketAddress expectedAddress = new InetSocketAddress("203.0.113.22", 7788);
-
- checkJobManagerAddress(config, expectedAddress.getHostName(), expectedAddress.getPort());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 1c1f4f4..b0a308d 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
@@ -30,8 +30,6 @@ import org.mockito.Mockito;
import java.util.Collections;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.isNull;
import static org.mockito.Matchers.notNull;
@@ -57,8 +55,7 @@ public class CliFrontendCancelTest extends TestLogger {
final ClusterClient clusterClient = createClusterClient();
MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- int retCode = testFrontend.cancel(parameters);
- assertEquals(0, retCode);
+ testFrontend.cancel(parameters);
Mockito.verify(clusterClient, times(1)).cancel(any(JobID.class));
}
@@ -71,8 +68,6 @@ public class CliFrontendCancelTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed.");
}
@Test(expected = CliArgsException.class)
@@ -83,8 +78,6 @@ public class CliFrontendCancelTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed with CliArgsException.");
}
/**
@@ -99,7 +92,7 @@ public class CliFrontendCancelTest extends TestLogger {
String[] parameters = { "-s", jid.toString() };
final ClusterClient clusterClient = createClusterClient();
MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- assertEquals(0, testFrontend.cancel(parameters));
+ testFrontend.cancel(parameters);
Mockito.verify(clusterClient, times(1))
.cancelWithSavepoint(any(JobID.class), isNull(String.class));
@@ -112,7 +105,7 @@ public class CliFrontendCancelTest extends TestLogger {
String[] parameters = { "-s", "targetDirectory", jid.toString() };
final ClusterClient clusterClient = createClusterClient();
MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- assertEquals(0, testFrontend.cancel(parameters));
+ testFrontend.cancel(parameters);
Mockito.verify(clusterClient, times(1))
.cancelWithSavepoint(any(JobID.class), notNull(String.class));
@@ -128,8 +121,6 @@ public class CliFrontendCancelTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed.");
}
@Test(expected = CliArgsException.class)
@@ -141,8 +132,6 @@ public class CliFrontendCancelTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed.");
}
private static ClusterClient createClusterClient() throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 3f1de55..fa212cb 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
@@ -47,8 +47,6 @@ public class CliFrontendInfoTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed with CliArgsException");
}
@Test(expected = CliArgsException.class)
@@ -59,8 +57,6 @@ public class CliFrontendInfoTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.cancel(parameters);
-
- fail("Should have failed with CliArgsException");
}
@Test
@@ -73,8 +69,7 @@ public class CliFrontendInfoTest extends TestLogger {
new Configuration(),
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
- int retCode = testFrontend.info(parameters);
- assertTrue(retCode == 0);
+ testFrontend.info(parameters);
assertTrue(buffer.toString().contains("\"parallelism\": \"1\""));
}
finally {
@@ -91,8 +86,7 @@ public class CliFrontendInfoTest extends TestLogger {
new Configuration(),
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
- int retCode = testFrontend.info(parameters);
- assertTrue(retCode == 0);
+ testFrontend.info(parameters);
assertTrue(buffer.toString().contains("\"parallelism\": \"17\""));
}
catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 8e10233..6639e25 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
@@ -30,8 +30,6 @@ import org.mockito.Mockito;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
@@ -53,8 +51,7 @@ public class CliFrontendListTest extends TestLogger {
String[] parameters = {"-r", "-s"};
ClusterClient clusterClient = createClusterClient();
MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- int retCode = testFrontend.list(parameters);
- assertTrue(retCode == 0);
+ testFrontend.list(parameters);
Mockito.verify(clusterClient, times(1))
.listJobs();
}
@@ -68,8 +65,6 @@ public class CliFrontendListTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.list(parameters);
-
- fail("Should have failed with an CliArgsException.");
}
private static ClusterClient createClusterClient() throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 26f26ff..a862dc1 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
@@ -193,7 +193,7 @@ public class CliFrontendPackageProgramTest extends TestLogger {
@Test(expected = CliArgsException.class)
public void testNoJarNoArgumentsAtAll() throws Exception {
- assertTrue(frontend.run(new String[0]) != 0);
+ frontend.run(new String[0]);
fail("Should have failed.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 bc55e2d..fa1c76c 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
@@ -33,7 +33,6 @@ import static org.apache.flink.client.cli.CliFrontendTestUtils.getTestJarPath;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
/**
* Tests for the RUN command.
@@ -51,28 +50,28 @@ public class CliFrontendRunTest {
{
String[] parameters = {"-v", getTestJarPath()};
RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(1, true, false);
- assertEquals(0, testFrontend.run(parameters));
+ testFrontend.run(parameters);
}
// test configure parallelism
{
String[] parameters = {"-v", "-p", "42", getTestJarPath()};
RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(42, true, false);
- assertEquals(0, testFrontend.run(parameters));
+ testFrontend.run(parameters);
}
// test configure sysout logging
{
String[] parameters = {"-p", "2", "-q", getTestJarPath()};
RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(2, false, false);
- assertEquals(0, testFrontend.run(parameters));
+ testFrontend.run(parameters);
}
// test detached mode
{
String[] parameters = {"-p", "2", "-d", getTestJarPath()};
RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(2, true, true);
- assertEquals(0, testFrontend.run(parameters));
+ testFrontend.run(parameters);
}
// test configure savepoint path (no ignore flag)
@@ -98,7 +97,7 @@ public class CliFrontendRunTest {
// test jar arguments
{
String[] parameters =
- {"-m", "localhost:6123", getTestJarPath(), "-arg1", "value1", "justavalue", "--arg2", "value2"};
+ { getTestJarPath(), "-arg1", "value1", "justavalue", "--arg2", "value2"};
RunOptions options = CliFrontendParser.parseRunCommand(parameters);
assertEquals("-arg1", options.getProgramArgs()[0]);
assertEquals("value1", options.getProgramArgs()[1]);
@@ -117,8 +116,6 @@ public class CliFrontendRunTest {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.run(parameters);
-
- fail("Should have failed.");
}
@Test(expected = CliArgsException.class)
@@ -130,8 +127,6 @@ public class CliFrontendRunTest {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.run(parameters);
-
- fail("Should have failed.");
}
@Test(expected = CliArgsException.class)
@@ -143,8 +138,6 @@ public class CliFrontendRunTest {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.run(parameters);
-
- fail("Should have failed.");
}
// --------------------------------------------------------------------------------------------
@@ -166,11 +159,10 @@ public class CliFrontendRunTest {
}
@Override
- protected int executeProgram(PackagedProgram program, ClusterClient client, int parallelism) {
+ protected void executeProgram(PackagedProgram program, ClusterClient client, int parallelism) {
assertEquals(isDetached, client.isDetached());
assertEquals(sysoutLogging, client.getPrintStatusDuringExecution());
assertEquals(expectedParallelism, parallelism);
- return 0;
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java
index f8b62cd..7b31a3e 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
@@ -45,7 +46,6 @@ import java.util.function.BiFunction;
import java.util.zip.ZipOutputStream;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -88,9 +88,8 @@ public class CliFrontendSavepointTest extends TestLogger {
MockedCliFrontend frontend = new MockedCliFrontend(clusterClient);
String[] parameters = { jobId.toString() };
- int returnCode = frontend.savepoint(parameters);
+ frontend.savepoint(parameters);
- assertEquals(0, returnCode);
verify(clusterClient, times(1))
.triggerSavepoint(eq(jobId), isNull(String.class));
@@ -120,8 +119,9 @@ public class CliFrontendSavepointTest extends TestLogger {
try {
frontend.savepoint(parameters);
- fail("This should have failed.");
- } catch (Exception e) {
+
+ fail("Savepoint should have failed.");
+ } catch (FlinkException e) {
assertTrue(ExceptionUtils.findThrowableWithMessage(e, expectedTestException).isPresent());
}
}
@@ -141,7 +141,6 @@ public class CliFrontendSavepointTest extends TestLogger {
new TestingHighAvailabilityServices()));
String[] parameters = { "invalid job id" };
-
try {
frontend.savepoint(parameters);
fail("Should have failed.");
@@ -172,9 +171,8 @@ public class CliFrontendSavepointTest extends TestLogger {
MockedCliFrontend frontend = new MockedCliFrontend(clusterClient);
String[] parameters = { jobId.toString(), savepointDirectory };
- int returnCode = frontend.savepoint(parameters);
+ frontend.savepoint(parameters);
- assertEquals(0, returnCode);
verify(clusterClient, times(1))
.triggerSavepoint(eq(jobId), eq(savepointDirectory));
@@ -269,11 +267,13 @@ public class CliFrontendSavepointTest extends TestLogger {
String[] parameters = { "-d", savepointPath };
- int returnCode = frontend.savepoint(parameters);
-
- assertNotEquals(0, returnCode);
+ try {
+ frontend.savepoint(parameters);
- assertTrue(buffer.toString().contains(testException.getMessage()));
+ fail("Savepoint should have failed.");
+ } catch (Exception e) {
+ assertTrue(ExceptionUtils.findThrowableWithMessage(e, testException.getMessage()).isPresent());
+ }
}
finally {
clusterClient.shutdown();
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 fc873cc..fbbc739 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
@@ -32,7 +32,6 @@ import org.mockito.Mockito;
import java.util.Collections;
import static org.apache.flink.client.cli.CliFrontendTestUtils.pipeSystemOutToNull;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
@@ -60,8 +59,7 @@ public class CliFrontendStopTest extends TestLogger {
final ClusterClient clusterClient = createClusterClient(false);
MockedCliFrontend testFrontend = new MockedCliFrontend(clusterClient);
- int retCode = testFrontend.stop(parameters);
- assertEquals(0, retCode);
+ testFrontend.stop(parameters);
Mockito.verify(clusterClient, times(1)).stop(any(JobID.class));
}
@@ -76,8 +74,6 @@ public class CliFrontendStopTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.stop(parameters);
-
- fail("Should have failed.");
}
@Test(expected = CliArgsException.class)
@@ -89,8 +85,6 @@ public class CliFrontendStopTest extends TestLogger {
Collections.singletonList(new DefaultCLI()),
CliFrontendTestUtils.getConfigDir());
testFrontend.stop(parameters);
-
- fail("Should have failed.");
}
@Test(expected = TestException.class)
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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
new file mode 100644
index 0000000..3f05f90
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/DefaultCLITest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.client.cli;
+
+import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.commons.cli.CommandLine;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Tests for the {@link DefaultCLI}.
+ */
+public class DefaultCLITest extends TestLogger {
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ /**
+ * Tests that the configuration is properly passed via the DefaultCLI to the
+ * created ClusterDescriptor.
+ */
+ @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 String localhost = "localhost";
+ final int port = 1234;
+ final Configuration configuration = new Configuration();
+
+ configuration.setString(JobManagerOptions.ADDRESS, localhost);
+ configuration.setInteger(JobManagerOptions.PORT, port);
+
+ final InetSocketAddress expectedAddress = new InetSocketAddress(localhost, port);
+
+ final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
+
+ final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(
+ configuration,
+ commandLine));
+
+ Assert.assertEquals(expectedAddress, clusterClient.getJobManagerAddress());
+ }
+
+ /**
+ * Tests that command line options override the configuration settings.
+ */
+ @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();
+
+ configuration.setString(JobManagerOptions.ADDRESS, localhost);
+ configuration.setInteger(JobManagerOptions.PORT, port);
+
+ final ClusterDescriptor<?> clusterDescriptor = defaultCLI.createClusterDescriptor(
+ configuration,
+ configurationDirectory,
+ commandLine);
+
+ final ClusterClient clusterClient = clusterDescriptor.retrieve(defaultCLI.getClusterId(
+ configuration,
+ commandLine));
+
+ final InetSocketAddress expectedAddress = new InetSocketAddress(manualHostname, manualPort);
+
+ Assert.assertEquals(expectedAddress, clusterClient.getJobManagerAddress());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
new file mode 100644
index 0000000..e46957b
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
@@ -0,0 +1,64 @@
+/*
+ * 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.client.cli.util;
+
+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.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * Dummy {@link ClusterDescriptor} implementation for testing purposes.
+ *
+ * @param <C> type of the returned {@link ClusterClient}
+ */
+public class DummyClusterDescriptor<C extends ClusterClient> implements ClusterDescriptor<C> {
+
+ private final C clusterClient;
+
+ public DummyClusterDescriptor(C clusterClient) {
+ this.clusterClient = Preconditions.checkNotNull(clusterClient);
+ }
+
+ @Override
+ public String getClusterDescription() {
+ return "DummyClusterDescriptor";
+ }
+
+ @Override
+ public C retrieve(String applicationID) throws UnsupportedOperationException {
+ return clusterClient;
+ }
+
+ @Override
+ public C deploySessionCluster(ClusterSpecification clusterSpecification) throws UnsupportedOperationException {
+ return clusterClient;
+ }
+
+ @Override
+ public C deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) {
+ return clusterClient;
+ }
+
+ @Override
+ public void close() {
+ // nothing to do
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2f1ba92/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 1bbafae..f9c2bfa 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
@@ -19,6 +19,8 @@
package org.apache.flink.client.cli.util;
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;
@@ -26,8 +28,7 @@ import org.apache.flink.util.Preconditions;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
-import java.net.URL;
-import java.util.List;
+import javax.annotation.Nullable;
/**
* Dummy implementation of the {@link CustomCommandLine} for testing purposes.
@@ -62,12 +63,21 @@ public class DummyCustomCommandLine<T extends ClusterClient> implements CustomCo
}
@Override
- public T retrieveCluster(CommandLine commandLine, Configuration config, String configurationDirectory) throws UnsupportedOperationException {
- return clusterClient;
+ public ClusterDescriptor<T> createClusterDescriptor(
+ Configuration configuration,
+ String configurationDirectory,
+ CommandLine commandLine) {
+ return new DummyClusterDescriptor<>(clusterClient);
}
@Override
- public T createCluster(String applicationName, CommandLine commandLine, Configuration config, String configurationDirectory, List<URL> userJarFiles) throws Exception {
- return clusterClient;
+ @Nullable
+ public String getClusterId(Configuration configuration, CommandLine commandLine) {
+ return "dummy";
+ }
+
+ @Override
+ public ClusterSpecification getClusterSpecification(Configuration configuration, CommandLine commandLine) {
+ return new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
}
}