[FLINK-2298] Add option to pass a custom name for Flink on YARN

This closes apache#876

docs/setup/yarn_setup.md

@@ -100,11 +100,13 @@ Usage:
  -D <arg> Dynamic properties
  -d,--detached Start detached
  -jm,--jobManagerMemory <arg> Memory for JobManager Container [in MB]
+ -nm,--name Set a custom name for the application on YARN
  -q,--query Display available YARN resources (memory, cores)
  -qu,--queue <arg> Specify YARN queue.
  -s,--slots <arg> Number of slots per TaskManager
  -st,--streaming Start Flink in streaming mode
  -tm,--taskManagerMemory <arg> Memory per TaskManager Container [in MB]
+
 ~~~
 
 Please note that the Client requires the `YARN_CONF_DIR` or `HADOOP_CONF_DIR` environment variable to be set to read the YARN and HDFS configuration.

flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java

@@ -134,8 +134,6 @@ public class CliFrontend {
 
  private AbstractFlinkYarnCluster yarnCluster;
 
-
-
  /**
  *
  * @throws Exception Thrown if teh configuration directory was not found, the configuration could not
@@ -744,7 +742,7 @@ protected Client getClient(CommandLineOptions options, ClassLoader classLoader,
  // user wants to run Flink in YARN cluster.
  CommandLine commandLine = options.getCommandLine();
  AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser.getFlinkYarnSessionCli().createFlinkYarnClient(commandLine);
-
+ flinkYarnClient.setName("Flink Application: " + programName);
  if (flinkYarnClient == null) {
  throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
  }
@@ -763,7 +761,7 @@ protected Client getClient(CommandLineOptions options, ClassLoader classLoader,
  }
 
  try {
- yarnCluster = flinkYarnClient.deploy("Flink Application: " + programName);
+ yarnCluster = flinkYarnClient.deploy();
  yarnCluster.connectToCluster();
  }
  catch(Exception e) {

flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java

@@ -74,6 +74,7 @@ public class FlinkYarnSessionCli {
  private final Option SLOTS;
  private final Option DETACHED;
  private final Option STREAMING;
+ private final Option NAME;
 
  /**
  * Dynamic properties allow the user to specify additional configuration values with -D, such as
@@ -97,6 +98,7 @@ public FlinkYarnSessionCli(String shortPrefix, String longPrefix) {
  DYNAMIC_PROPERTIES = new Option(shortPrefix + "D", true, "Dynamic properties");
  DETACHED = new Option(shortPrefix + "d", longPrefix + "detached", false, "Start detached");
  STREAMING = new Option(shortPrefix + "st", longPrefix + "streaming", false, "Start Flink in streaming mode");
+ NAME = new Option(shortPrefix + "nm", longPrefix + "name", true, "Set a custom name for the application on YARN");
  }
 
  public AbstractFlinkYarnClient createFlinkYarnClient(CommandLine cmd) {
@@ -220,6 +222,9 @@ public boolean accept(File dir, String name) {
  if (cmd.hasOption(STREAMING.getOpt())) {
  flinkYarnClient.setStreamingMode(true);
  }
+ if(cmd.hasOption(NAME.getOpt())) {
+ flinkYarnClient.setName(cmd.getOptionValue(NAME.getOpt()));
+ }
  return flinkYarnClient;
  }
 
@@ -244,6 +249,7 @@ private void printUsage() {
  opt.addOption(DYNAMIC_PROPERTIES);
  opt.addOption(DETACHED);
  opt.addOption(STREAMING);
+ opt.addOption(NAME);
  formatter.printHelp(" ", opt);
  }
 
@@ -350,6 +356,7 @@ public void getYARNSessionCLIOptions(Options options) {
  options.addOption(DYNAMIC_PROPERTIES);
  options.addOption(DETACHED);
  options.addOption(STREAMING);
+ options.addOption(NAME);
  }
 
  public int run(String[] args) {
@@ -393,7 +400,7 @@ public int run(String[] args) {
 
 
  try {
- yarnCluster = flinkYarnClient.deploy(null);
+ yarnCluster = flinkYarnClient.deploy();
  // only connect to cluster if its not a detached session.
  if(!flinkYarnClient.isDetached()) {
  yarnCluster.connectToCluster();

flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnClient.java

@@ -115,9 +115,8 @@ public abstract class AbstractFlinkYarnClient {
  /**
  * Trigger the deployment to YARN.
  *
- * @param clusterName Name to be shown in the YARN resource manager overview.
  */
- public abstract AbstractFlinkYarnCluster deploy(String clusterName) throws Exception;
+ public abstract AbstractFlinkYarnCluster deploy() throws Exception;
 
  /**
  * @param detachedMode If true, the Flink YARN client is non-blocking. That means it returns
@@ -138,4 +137,10 @@ public abstract class AbstractFlinkYarnClient {
  * @param streamingMode
  */
  public abstract void setStreamingMode(boolean streamingMode);
+
+ /**
+ * Set a name for the YARN application
+ * @param name
+ */
+ public abstract void setName(String name);
 }

flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java

@@ -121,6 +121,7 @@ public void testDetachedMode() {
  "-n", "1",
  "-jm", "768",
  "-tm", "1024",
+ "--name", "MyCustomName", // test setting a custom name
  "--detached"},
  "Flink JobManager is now running on", RunTypes.YARN_SESSION);
 
@@ -142,7 +143,9 @@ public void testDetachedMode() {
  yc.start();
  List<ApplicationReport> apps = yc.getApplications(EnumSet.of(YarnApplicationState.RUNNING));
  Assert.assertEquals(1, apps.size()); // Only one running
- ApplicationId id = apps.get(0).getApplicationId();
+ ApplicationReport app = apps.get(0);
+ Assert.assertEquals("MyCustomName", app.getName());
+ ApplicationId id = app.getApplicationId();
  yc.killApplication(id);
 
  while(yc.getApplications(EnumSet.of(YarnApplicationState.KILLED)).size() == 0) {
@@ -166,6 +169,7 @@ public void testTaskManagerFailure() {
  "-n", "1",
  "-jm", "768",
  "-tm", "1024",
+ "-nm", "customName",
  "-Dfancy-configuration-value=veryFancy",
  "-Dyarn.maximum-failed-containers=3"},
  "Number of connected TaskManagers changed to 1. Slots available: 1",
@@ -180,7 +184,9 @@ public void testTaskManagerFailure() {
  yc.start();
  List<ApplicationReport> apps = yc.getApplications(EnumSet.of(YarnApplicationState.RUNNING));
  Assert.assertEquals(1, apps.size()); // Only one running
- String url = apps.get(0).getTrackingUrl();
+ ApplicationReport app = apps.get(0);
+ Assert.assertEquals("customName", app.getName());
+ String url = app.getTrackingUrl();
  if(!url.endsWith("/")) {
  url += "/";
  }
@@ -615,7 +621,7 @@ public void testJavaAPI() {
  // deploy
  AbstractFlinkYarnCluster yarnCluster = null;
  try {
- yarnCluster = flinkYarnClient.deploy(null);
+ yarnCluster = flinkYarnClient.deploy();
  yarnCluster.connectToCluster();
  } catch (Exception e) {
  System.err.println("Error while deploying YARN cluster: "+e.getMessage());

flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java

@@ -143,6 +143,7 @@ public class FlinkYarnClient extends AbstractFlinkYarnClient {
  private boolean detached;
  private boolean streamingMode;
 
+ private String customName = null;
 
  public FlinkYarnClient() {
  conf = new YarnConfiguration();
@@ -314,7 +315,7 @@ public boolean isDetached() {
  return detached;
  }
 
- public AbstractFlinkYarnCluster deploy(final String clusterName) throws Exception {
+ public AbstractFlinkYarnCluster deploy() throws Exception {
 
  UserGroupInformation.setConfiguration(conf);
  UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
@@ -327,11 +328,11 @@ public AbstractFlinkYarnCluster deploy(final String clusterName) throws Exceptio
  return ugi.doAs(new PrivilegedExceptionAction<AbstractFlinkYarnCluster>() {
  @Override
  public AbstractFlinkYarnCluster run() throws Exception {
- return deployInternal(clusterName);
+ return deployInternal();
  }
  });
  } else {
- return deployInternal(clusterName);
+ return deployInternal();
  }
  }
 
@@ -341,7 +342,7 @@ public AbstractFlinkYarnCluster run() throws Exception {
  * This method will block until the ApplicationMaster/JobManager have been
  * deployed on YARN.
  */
- protected AbstractFlinkYarnCluster deployInternal(String clusterName) throws Exception {
+ protected AbstractFlinkYarnCluster deployInternal() throws Exception {
  isReadyForDepoyment();
 
  LOG.info("Using values:");
@@ -591,14 +592,17 @@ protected AbstractFlinkYarnCluster deployInternal(String clusterName) throws Exc
  capability.setMemory(jobManagerMemoryMb);
  capability.setVirtualCores(1);
 
- if(clusterName == null) {
- clusterName = "Flink session with "+taskManagerCount+" TaskManagers";
- }
- if(detached) {
- clusterName += " (detached)";
+ String name;
+ if(customName == null) {
+ name = "Flink session with "+taskManagerCount+" TaskManagers";
+ if(detached) {
+ name += " (detached)";
+ }
+ } else {
+ name = customName;
  }
 
- appContext.setApplicationName(clusterName); // application name
+ appContext.setApplicationName(name); // application name
  appContext.setApplicationType("Apache Flink");
  appContext.setAMContainerSpec(amContainer);
  appContext.setResource(capability);
@@ -734,6 +738,14 @@ public void setStreamingMode(boolean streamingMode) {
  this.streamingMode = streamingMode;
  }
 
+ @Override
+ public void setName(String name) {
+ if(name == null) {
+ throw new IllegalArgumentException("The passed name is null");
+ }
+ customName = name;
+ }
+
  public static class YarnDeploymentException extends RuntimeException {
  public YarnDeploymentException() {
  }