You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by mxm <gi...@git.apache.org> on 2016/06/08 18:20:12 UTC
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
GitHub user mxm opened a pull request:
https://github.com/apache/flink/pull/2085
[FLINK-3937] programmatic resuming of clusters
These changes are based on #1978 and #2034. More specifically, they port resuming of running Yarn clusters from #2034 to the refactoring of #1978. The abstraction in place enables us to plug in other cluster frameworks in the future.
- integrates with and extends the refactoring of FLINK-3667
- enables to resume from Yarn properties or Yarn application id
- introduces additional StandaloneClusterDescriptor
- introduces DefaultCLI to get rid of standalone mode switches in CliFrontend
- various fixes and improvements
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mxm/flink FLINK-3937
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2085.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2085
----
commit b144e64758a95bcac33bd0ac91ab7eefaf4040e9
Author: Maximilian Michels <mx...@apache.org>
Date: 2016-04-22T17:52:54Z
[FLINK-3667] refactor client communication classes
- ClusterDescriptor: base interface for cluster deployment descriptors
- ClusterDescriptor: YarnClusterDescriptor
- ClusterClient: base class for ClusterClients, handles lifecycle of cluster
- ClusterClient: shares configuration with the implementations
- ClusterClient: StandaloneClusterClient, YarnClusterClient
- ClusterClient: remove run methods and enable detached mode via flag
- CliFrontend: remove all Yarn specific logic
- CliFrontend: remove all cluster setup logic
- CustomCommandLine: interface for other cluster implementations
- Customcommandline: enables creation of new cluster or resuming from existing
- Yarn: move Yarn classes and functionality to the yarn module (yarn
properties, yarn interfaces)
- Yarn: improve reliability of cluster startup
- Yarn Tests: only disable parallel execution of ITCases
commit 73524c89854f04ac41f0c288d9ebf8ef5efe628b
Author: Sebastian Klemke <se...@researchgate.net>
Date: 2016-05-25T12:28:59Z
[FLINK-3937] implement -yid option to Flink CLI
- enables to use list, savepoint, cancel and stop subcommands
- adapt FlinkYarnSessionCli to also accept YARN application Id to attach to
- update documentation
commit 1db8c97c39c2bf071db018c1ca505409c847a30b
Author: Maximilian Michels <mx...@apache.org>
Date: 2016-06-01T10:45:52Z
[FLINK-3863] Yarn Cluster shutdown may fail if leader changed recently
commit 1a154fb12474a8630cce7e764d72398513055887
Author: Maximilian Michels <mx...@apache.org>
Date: 2016-06-02T14:28:51Z
[FLINK-3937] programmatic resuming of clusters
- integrates with and extends the refactoring of FLINK-3667
- enables to resume from Yarn properties or Yarn application id
- introduces additional StandaloneClusterDescriptor
- introduces DefaultCLI to get rid of standalone mode switches in CliFrontend
- various fixes and improvements
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r67003363
--- Diff: docs/apis/cli.md ---
@@ -252,6 +256,12 @@ Action "list" lists running and scheduled programs.
configuration.
-r,--running Show only running programs and their JobIDs
-s,--scheduled Show only scheduled programs and their JobIDs
+ Additional arguments if -m yarn-cluster is set:
+ -yid <yarnApplicationId> YARN application ID of Flink YARN session to
+ connect to. Must not be set if JobManager HA
+ is used. In this case, JobManager RPC
+ location is automatically retrieved from
+ Zookeeper.
--- End diff --
The code no longer waits for TaskManagers when the cluster is resumed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
Thanks a lot for adding this feature. I looked through the code and tried it out on a cluster.
Once my concerns are addressed (either through changes or comments) we can merge the change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66799100
--- Diff: docs/apis/cli.md ---
@@ -105,6 +105,10 @@ The command line can be used to
./bin/flink list -r
+- List running Flink jobs inside Flink YARN session:
+
+ ./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r
--- End diff --
Are there any tests for this functionality?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66797696
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java ---
@@ -300,18 +309,82 @@ public static boolean allocateResource(int[] nodeManagers, int toAllocate) {
return false;
}
- @Override
public void setDetachedMode(boolean detachedMode) {
this.detached = detachedMode;
}
- @Override
- public boolean isDetached() {
+ public boolean isDetachedMode() {
return detached;
}
+
+ /**
+ * Gets a Hadoop Yarn client
+ * @return Returns a YarnClient which has to be shutdown manually
+ */
+ private static YarnClient getYarnClient(Configuration conf) {
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ return yarnClient;
+ }
+
+ /**
+ * Retrieves the Yarn application and cluster from the config
+ * @param config The config with entries to retrieve the cluster
+ * @return YarnClusterClient
+ * @deprecated This should be removed in the future
+ */
+ public YarnClusterClient retrieveFromConfig(org.apache.flink.configuration.Configuration config) throws Exception {
+ String jobManagerHost = config.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int jobManagerPort = config.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+
+ if (jobManagerHost != null && jobManagerPort != -1) {
+
+ YarnClient yarnClient = getYarnClient(conf);
+ List<ApplicationReport> applicationReports = yarnClient.getApplications();
+ for (ApplicationReport report : applicationReports) {
+ if (report.getHost().equals(jobManagerHost) && report.getRpcPort() == jobManagerPort) {
+ LOG.info("Found application '{}' " +
+ "with JobManager host name '{}' and port '{}' from Yarn properties file.",
+ report.getApplicationId(), jobManagerHost, jobManagerPort);
+ return retrieve(report.getApplicationId().toString());
+ }
+ }
+
--- End diff --
I wonder if it makes sense to log here that we were unable to find the application id?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
Detached per job submission is not working (`./bin/flink run -m yarn-cluster -yd -yn 1 ./examples/batch/WordCount.jar `)
```
2016-06-13 07:28:22,423 INFO org.apache.flink.yarn.ApplicationClient - Trying to register at JobManager akka.tcp://flink@10.0.2.15:45521/user/jobmanager.
2016-06-13 07:28:22,829 INFO org.apache.flink.yarn.ApplicationClient - Successfully registered at the ResourceManager using JobManager Actor[akka.tcp://flink@10.0.2.15:45521/user/jobmanager#1868998090]
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
TaskManager status (0/1)
All TaskManagers are connected
Cluster started
Using address /10.0.2.15:45521 to connect to JobManager.
JobManager web interface address http://quickstart.cloudera:8088/proxy/application_1447844011707_0039/
Starting execution of program
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
------------------------------------------------------------
The program finished with the following exception:
Job was submitted in detached mode. Results of job execution, such as accumulators, runtime, job id etc. are not available. Please make sure your program doesn't call an eager execution function [collect, print, printToErr, count].
org.apache.flink.client.program.DetachedEnvironment$DetachedJobExecutionResult.getAccumulatorResult(DetachedEnvironment.java:103)
org.apache.flink.api.java.DataSet.collect(DataSet.java:412)
org.apache.flink.api.java.DataSet.print(DataSet.java:1605)
org.apache.flink.examples.java.wordcount.WordCount.main(WordCount.java:92)
2016-06-13 07:28:27,206 INFO org.apache.flink.yarn.YarnClusterClient - Disconnecting YarnClusterClient from ApplicationMaster
2016-06-13 07:28:27,208 INFO org.apache.flink.yarn.ApplicationClient - Stopped Application client.
2016-06-13 07:28:27,208 INFO org.apache.flink.yarn.ApplicationClient - Disconnect from JobManager Actor[akka.tcp://flink@10.0.2.15:45521/user/jobmanager#1868998090].
2016-06-13 07:28:27,307 INFO org.apache.flink.yarn.YarnClusterClient - Application application_1447844011707_0039 finished with state RUNNING and final state UNDEFINED at 0
[cloudera@quickstart build-target]
```
In the JobManager logs, there is no sign of a job being submitted.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
I'll probably continue reviewing your work next Monday.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66795170
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java ---
@@ -415,62 +513,83 @@ public int run(String[] args) {
}
System.out.println(description);
return 0;
+ } else if (cmd.hasOption(APPLICATION_ID.getOpt())) { // TODO RM
--- End diff --
What is this TODO about?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805499
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java ---
@@ -300,18 +309,82 @@ public static boolean allocateResource(int[] nodeManagers, int toAllocate) {
return false;
}
- @Override
public void setDetachedMode(boolean detachedMode) {
this.detached = detachedMode;
}
- @Override
- public boolean isDetached() {
+ public boolean isDetachedMode() {
return detached;
}
+
+ /**
+ * Gets a Hadoop Yarn client
+ * @return Returns a YarnClient which has to be shutdown manually
+ */
+ private static YarnClient getYarnClient(Configuration conf) {
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ return yarnClient;
+ }
+
+ /**
+ * Retrieves the Yarn application and cluster from the config
+ * @param config The config with entries to retrieve the cluster
+ * @return YarnClusterClient
+ * @deprecated This should be removed in the future
+ */
+ public YarnClusterClient retrieveFromConfig(org.apache.flink.configuration.Configuration config) throws Exception {
+ String jobManagerHost = config.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int jobManagerPort = config.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+
+ if (jobManagerHost != null && jobManagerPort != -1) {
+
+ YarnClient yarnClient = getYarnClient(conf);
+ List<ApplicationReport> applicationReports = yarnClient.getApplications();
+ for (ApplicationReport report : applicationReports) {
+ if (report.getHost().equals(jobManagerHost) && report.getRpcPort() == jobManagerPort) {
+ LOG.info("Found application '{}' " +
+ "with JobManager host name '{}' and port '{}' from Yarn properties file.",
+ report.getApplicationId(), jobManagerHost, jobManagerPort);
+ return retrieve(report.getApplicationId().toString());
+ }
+ }
+
+ }
+ return null;
--- End diff --
Same as above. Fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the issue:
https://github.com/apache/flink/pull/2085
I've addressed your comments in another commit. I'd be delighted if you could try out the changes again!
- remove legacy code from CliFrontend
- change activation code of CustomCommandLine interface
- use checked exceptions to signal supported operations
- remove all checked exceptions of type Exception
- fix proper handling of yarn properties file
- fix logging and reduce verbosity of per-job clusters
- print 'id' argument in YarnSessionCli
- minor renaming of methods names
- improve documentation
- deprecate streaming option
- deprecate -m 'yarn-cluster'
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66800124
--- Diff: docs/apis/cli.md ---
@@ -252,6 +256,12 @@ Action "list" lists running and scheduled programs.
configuration.
-r,--running Show only running programs and their JobIDs
-s,--scheduled Show only scheduled programs and their JobIDs
+ Additional arguments if -m yarn-cluster is set:
+ -yid <yarnApplicationId> YARN application ID of Flink YARN session to
+ connect to. Must not be set if JobManager HA
+ is used. In this case, JobManager RPC
+ location is automatically retrieved from
+ Zookeeper.
--- End diff --
I tried this, but I wonder why its not logging that the job has been cancelled.
It logs that "All TaskManagers are connected", I don't think this message is relevant when cancelling a job.
```
[cloudera@quickstart build-target]$ ./bin/flink cancel -m yarn-cluster -yid application_1447844011707_0038 b9b8f76616073d09c596545a3cda978f
2016-06-13 07:22:21,355 INFO org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at quickstart.cloudera/10.0.2.15:8032
2016-06-13 07:22:22,277 INFO org.apache.flink.yarn.YarnClusterClient - Start application client.
2016-06-13 07:22:22,288 INFO org.apache.flink.yarn.ApplicationClient - Notification about new leader address akka.tcp://flink@10.0.2.15:51747/user/jobmanager with session ID null.
2016-06-13 07:22:22,290 INFO org.apache.flink.yarn.ApplicationClient - Received address of new leader akka.tcp://flink@10.0.2.15:51747/user/jobmanager with session ID null.
2016-06-13 07:22:22,290 INFO org.apache.flink.yarn.ApplicationClient - Disconnect from JobManager null.
Waiting until all TaskManagers have connected
2016-06-13 07:22:22,297 INFO org.apache.flink.yarn.ApplicationClient - Trying to register at JobManager akka.tcp://flink@10.0.2.15:51747/user/jobmanager.
No status updates from the YARN cluster received so far. Waiting ...
2016-06-13 07:22:22,542 INFO org.apache.flink.yarn.ApplicationClient - Successfully registered at the ResourceManager using JobManager Actor[akka.tcp://flink@10.0.2.15:51747/user/jobmanager#1733798764]
All TaskManagers are connected
2016-06-13 07:22:22,945 INFO org.apache.flink.yarn.YarnClusterClient - Shutting down YarnClusterClient from the client shutdown hook
2016-06-13 07:22:22,945 INFO org.apache.flink.yarn.YarnClusterClient - Disconnecting YarnClusterClient from ApplicationMaster
2016-06-13 07:22:22,947 INFO org.apache.flink.yarn.ApplicationClient - Stopped Application client.
2016-06-13 07:22:22,947 INFO org.apache.flink.yarn.ApplicationClient - Disconnect from JobManager Actor[akka.tcp://flink@10.0.2.15:51747/user/jobmanager#1733798764].
2016-06-13 07:22:23,056 INFO org.apache.flink.yarn.YarnClusterClient - Application application_1447844011707_0038 finished with state RUNNING and final state UNDEFINED at 0
[cloudera@quickstart build-target]$
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r67018443
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java ---
@@ -211,18 +185,45 @@ public void run() {
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
isConnected = true;
+
+ logAndSysout("Waiting until all TaskManagers have connected");
+
+ while(true) {
--- End diff --
I agree that the functionality is the same.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the issue:
https://github.com/apache/flink/pull/2085
Rebased to the latest master and refined the tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66795439
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java ---
@@ -108,24 +131,83 @@ public FlinkYarnSessionCli(String shortPrefix, String longPrefix, boolean accept
NAME = new Option(shortPrefix + "nm", longPrefix + "name", true, "Set a custom name for the application on YARN");
}
+
/**
- * Creates a new Yarn Client.
- * @param cmd the command line to parse options from
- * @return an instance of the client or null if there was an error
+ * Resumes from a Flink Yarn properties file
+ * @param flinkConfiguration The flink configuration
+ * @return True if the properties were loaded, false otherwise
*/
- public AbstractFlinkYarnClient createFlinkYarnClient(CommandLine cmd) {
+ private boolean resumeFromYarnProperties(Configuration flinkConfiguration) {
+ // load the YARN properties
+ File propertiesFile = new File(getYarnPropertiesLocation(flinkConfiguration));
+ if (!propertiesFile.exists()) {
+ return false;
+ }
+
+ logAndSysout("Found YARN properties file " + propertiesFile.getAbsolutePath());
+
+ Properties yarnProperties = new Properties();
+ try {
+ try (InputStream is = new FileInputStream(propertiesFile)) {
+ yarnProperties.load(is);
+ }
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Cannot read the YARN properties file", e);
+ }
+
+ // configure the default parallelism from YARN
+ String propParallelism = yarnProperties.getProperty(YARN_PROPERTIES_PARALLELISM);
+ if (propParallelism != null) { // maybe the property is not set
+ try {
+ int parallelism = Integer.parseInt(propParallelism);
+ flinkConfiguration.setInteger(ConfigConstants.DEFAULT_PARALLELISM_KEY, parallelism);
+
+ logAndSysout("YARN properties set default parallelism to " + parallelism);
+ }
+ catch (NumberFormatException e) {
+ throw new RuntimeException("Error while parsing the YARN properties: " +
+ "Property " + YARN_PROPERTIES_PARALLELISM + " is not an integer.");
+ }
+ }
+
+ // get the JobManager address from the YARN properties
+ String address = yarnProperties.getProperty(YARN_PROPERTIES_JOBMANAGER_KEY);
+ InetSocketAddress jobManagerAddress;
+ if (address != null) {
+ try {
+ jobManagerAddress = ClientUtils.parseHostPortAddress(address);
+ // store address in config from where it is retrieved by the retrieval service
+ CliFrontend.writeJobManagerAddressToConfig(flinkConfiguration, jobManagerAddress);
+ }
+ catch (Exception e) {
+ throw new RuntimeException("YARN properties contain an invalid entry for JobManager address.", e);
+ }
+
+ logAndSysout("Using JobManager address from YARN properties " + jobManagerAddress);
+ }
- AbstractFlinkYarnClient flinkYarnClient = getFlinkYarnClient();
- if (flinkYarnClient == null) {
- return null;
+ // handle the YARN client's dynamic properties
+ String dynamicPropertiesEncoded = yarnProperties.getProperty(YARN_PROPERTIES_DYNAMIC_PROPERTIES_STRING);
+ Map<String, String> dynamicProperties = getDynamicProperties(dynamicPropertiesEncoded);
+ for (Map.Entry<String, String> dynamicProperty : dynamicProperties.entrySet()) {
+ flinkConfiguration.setString(dynamicProperty.getKey(), dynamicProperty.getValue());
}
+ return true;
+ }
+
+ public YarnClusterDescriptor createDescriptor(String defaultApplicationName, CommandLine cmd) {
+
+
+ YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor();
--- End diff --
A bit too many blank lines ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805417
--- Diff: docs/apis/cli.md ---
@@ -105,6 +105,10 @@ The command line can be used to
./bin/flink list -r
+- List running Flink jobs inside Flink YARN session:
+
+ ./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r
--- End diff --
Thank you, I'll look into this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805479
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java ---
@@ -211,18 +185,45 @@ public void run() {
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
isConnected = true;
+
+ logAndSysout("Waiting until all TaskManagers have connected");
+
+ while(true) {
--- End diff --
Yes, I wanted to bring up the Yarn session only once the cluster is ready. It is a semantic change but IMHO transparent to the user. There is not disadvantage from waiting until the cluster is ready. Ultimately, it would be nice to get rid of all waiting but we're not quite there yet.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805953
--- Diff: flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java ---
@@ -0,0 +1,65 @@
+/*
+ * 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.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+
+
+/**
+ * Custom command-line interface to load hooks for the command-line interface.
+ */
+public interface CustomCommandLine<ClusterType extends ClusterClient> {
+
+ /**
+ * Returns a unique identifier for this custom command-line.
+ * @return An unique identifier string
+ */
+ String getIdentifier();
+
+ /**
+ * Adds custom options to the existing run options.
+ * @param baseOptions The existing options.
+ */
+ void addRunOptions(Options baseOptions);
+
+ /**
+ * Adds custom options to the existing general options.
+ * @param baseOptions The existing options.
+ */
+ 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
+ * @return Client if a cluster could be retrieve, null otherwise
--- End diff --
Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the issue:
https://github.com/apache/flink/pull/2085
Merged. Thank you for your comments!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66626078
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -1275,33 +1071,16 @@ else if (new File(CONFIG_DIRECTORY_FALLBACK_2).exists()) {
return location;
}
- public static Map<String, String> getDynamicProperties(String dynamicPropertiesEncoded) {
- if (dynamicPropertiesEncoded != null && dynamicPropertiesEncoded.length() > 0) {
- Map<String, String> properties = new HashMap<>();
-
- String[] propertyLines = dynamicPropertiesEncoded.split(CliFrontend.YARN_DYNAMIC_PROPERTIES_SEPARATOR);
- for (String propLine : propertyLines) {
- if (propLine == null) {
- continue;
- }
-
- String[] kv = propLine.split("=");
- if (kv.length >= 2 && kv[0] != null && kv[1] != null && kv[0].length() > 0) {
- properties.put(kv[0], kv[1]);
- }
- }
- return properties;
- }
- else {
- return Collections.emptyMap();
- }
- }
- public static String getYarnPropertiesLocation(Configuration conf) {
- String defaultPropertiesFileLocation = System.getProperty("java.io.tmpdir");
- String currentUser = System.getProperty("user.name");
- String propertiesFileLocation = conf.getString(ConfigConstants.YARN_PROPERTIES_FILE_LOCATION, defaultPropertiesFileLocation);
-
- return propertiesFileLocation + File.separator + CliFrontend.YARN_PROPERTIES_FILE + currentUser;
+ /**
+ * Writes the given job manager address to the associated configuration object
+ *
+ * @param address Address to write to the configuration
+ * @param config The config to write to
+ */
+ public static void writeJobManagerAddressToConfig(Configuration config, InetSocketAddress address) {
--- End diff --
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
I tried your code again on my testing machine and everything that didn't work at my last tests is now fixed.
I also quickly scanned the newly added commits.
I think the pull request is good to merge.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66797580
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java ---
@@ -211,18 +185,45 @@ public void run() {
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
isConnected = true;
+
+ logAndSysout("Waiting until all TaskManagers have connected");
+
+ while(true) {
--- End diff --
It seems that this loop is now executed also for the YARN session. In the past, this was used only for "per job" yarn clusters, because there we needed to have all slots available to submit the job.
Now, this loop is "blocking" the YARN session CLI if not all TaskManager's are connecting. Its not a big deal, but a change in the semantics.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66794640
--- Diff: docs/apis/cli.md ---
@@ -105,6 +105,10 @@ The command line can be used to
./bin/flink list -r
+- List running Flink jobs inside Flink YARN session:
+
+ ./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r
--- End diff --
This is a similar thing:
```
[cloudera@quickstart build-target]$ ./bin/flink list -m yarn-cluster -yid obiouslyWrong -r
------------------------------------------------------------
The program finished with the following exception:
org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException: Could not retrieve the leader gateway
at org.apache.flink.runtime.util.LeaderRetrievalUtils.retrieveLeaderGateway(LeaderRetrievalUtils.java:127)
at org.apache.flink.client.CliFrontend.getJobManagerGateway(CliFrontend.java:844)
at org.apache.flink.client.CliFrontend.list(CliFrontend.java:378)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:986)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1034)
Caused by: java.util.concurrent.TimeoutException: Futures timed out after [10000 milliseconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at scala.concurrent.Await.result(package.scala)
at org.apache.flink.runtime.util.LeaderRetrievalUtils.retrieveLeaderGateway(LeaderRetrievalUtils.java:125)
... 4 more
[cloudera@quickstart build-target]$
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on the issue:
https://github.com/apache/flink/pull/2085
Thanks for the review! I've addressed almost all your comments and will ping you once the PR get updated. Some more comments:
> (1) When listing the yarn-session.sh cli options, I could not see the newly added argument
Fixed. The argument was not printed. The code now prints all available options.
>Starting a YARN session works. Also, connecting with a second YARN session works. However, if I'm stopping the session from the second session client, the first one does not notice this.
It would be good if all connected clients receive the messages, instead of the last one connected.
Yes, would be nice to have that feature but I think it is out of the scope of this PR.
>Detached per job submission is not working (./bin/flink run -m yarn-cluster -yd -yn 1 ./examples/batch/WordCount.jar)
You're running a WordCount with interactive job submission detached. That doesn't work because the plan can't be extracted in interactive programs (programs which use count/collect/print).
>While trying out your code, I noticed that the .yarn-properties file is not properly deleted, even though my previous yarn session was shutting down correctly:
I'll look into it. I know that this is also an issue in the current master. I had the properties file lingering around many times. I think it is not properly cleaned up in the ShutdownHook.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66797799
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java ---
@@ -300,18 +309,82 @@ public static boolean allocateResource(int[] nodeManagers, int toAllocate) {
return false;
}
- @Override
public void setDetachedMode(boolean detachedMode) {
this.detached = detachedMode;
}
- @Override
- public boolean isDetached() {
+ public boolean isDetachedMode() {
return detached;
}
+
+ /**
+ * Gets a Hadoop Yarn client
+ * @return Returns a YarnClient which has to be shutdown manually
+ */
+ private static YarnClient getYarnClient(Configuration conf) {
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ return yarnClient;
+ }
+
+ /**
+ * Retrieves the Yarn application and cluster from the config
+ * @param config The config with entries to retrieve the cluster
+ * @return YarnClusterClient
+ * @deprecated This should be removed in the future
+ */
+ public YarnClusterClient retrieveFromConfig(org.apache.flink.configuration.Configuration config) throws Exception {
+ String jobManagerHost = config.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int jobManagerPort = config.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+
+ if (jobManagerHost != null && jobManagerPort != -1) {
+
+ YarnClient yarnClient = getYarnClient(conf);
+ List<ApplicationReport> applicationReports = yarnClient.getApplications();
+ for (ApplicationReport report : applicationReports) {
+ if (report.getHost().equals(jobManagerHost) && report.getRpcPort() == jobManagerPort) {
+ LOG.info("Found application '{}' " +
+ "with JobManager host name '{}' and port '{}' from Yarn properties file.",
+ report.getApplicationId(), jobManagerHost, jobManagerPort);
+ return retrieve(report.getApplicationId().toString());
+ }
+ }
+
+ }
+ return null;
--- End diff --
Does it make sense to log that the config hasn't been set?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66626302
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -980,110 +845,41 @@ protected ActorGateway getJobManagerGateway(CommandLineOptions options) throws E
}
/**
- * Retrieves a {@link Client} object from the given command line options and other parameters.
+ * Retrieves a {@link ClusterClient} object from the given command line options and other parameters.
*
* @param options Command line options which contain JobManager address
* @param programName Program name
- * @param userParallelism Given user parallelism
* @throws Exception
*/
- protected Client getClient(
+ protected ClusterClient getClient(
CommandLineOptions options,
- String programName,
- int userParallelism,
- boolean detachedMode)
- throws Exception {
- InetSocketAddress jobManagerAddress;
- int maxSlots = -1;
+ String programName) throws Exception {
- if (YARN_DEPLOY_JOBMANAGER.equals(options.getJobManagerAddress())) {
- logAndSysout("YARN cluster mode detected. Switching Log4j output to console");
-
- // Default yarn application name to use, if nothing is specified on the command line
- String applicationName = "Flink Application: " + programName;
-
- // user wants to run Flink in YARN cluster.
- CommandLine commandLine = options.getCommandLine();
- AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser
- .getFlinkYarnSessionCli()
- .withDefaultApplicationName(applicationName)
- .createFlinkYarnClient(commandLine);
-
- if (flinkYarnClient == null) {
- throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
- }
-
- // in case the main detached mode wasn't set, we don't wanna overwrite the one loaded
- // from yarn options.
- if (detachedMode) {
- flinkYarnClient.setDetachedMode(true);
- }
-
- // the number of slots available from YARN:
- int yarnTmSlots = flinkYarnClient.getTaskManagerSlots();
- if (yarnTmSlots == -1) {
- yarnTmSlots = 1;
- }
- maxSlots = yarnTmSlots * flinkYarnClient.getTaskManagerCount();
- if (userParallelism != -1) {
- int slotsPerTM = userParallelism / flinkYarnClient.getTaskManagerCount();
- logAndSysout("The YARN cluster has " + maxSlots + " slots available, " +
- "but the user requested a parallelism of " + userParallelism + " on YARN. " +
- "Each of the " + flinkYarnClient.getTaskManagerCount() + " TaskManagers " +
- "will get "+slotsPerTM+" slots.");
- flinkYarnClient.setTaskManagerSlots(slotsPerTM);
- }
-
- try {
- yarnCluster = flinkYarnClient.deploy();
- yarnCluster.connectToCluster();
- }
- catch (Exception e) {
- throw new RuntimeException("Error deploying the YARN cluster", e);
- }
+ // Get the custom command-line (e.g. Standalone/Yarn/Mesos)
+ CustomCommandLine<?> activeCommandLine =
+ CliFrontendParser.getActiveCustomCommandLine(options.getJobManagerAddress());
- jobManagerAddress = yarnCluster.getJobManagerAddress();
- writeJobManagerAddressToConfig(jobManagerAddress);
-
- // overwrite the yarn client config (because the client parses the dynamic properties)
- this.config.addAll(flinkYarnClient.getFlinkConfiguration());
-
- logAndSysout("YARN cluster started");
- logAndSysout("JobManager web interface address " + yarnCluster.getWebInterfaceURL());
- logAndSysout("Waiting until all TaskManagers have connected");
-
- while(true) {
- GetClusterStatusResponse status = yarnCluster.getClusterStatus();
- if (status != null) {
- if (status.numRegisteredTaskManagers() < flinkYarnClient.getTaskManagerCount()) {
- logAndSysout("TaskManager status (" + status.numRegisteredTaskManagers() + "/"
- + flinkYarnClient.getTaskManagerCount() + ")");
- } else {
- logAndSysout("All TaskManagers are connected");
- break;
- }
- } else {
- logAndSysout("No status updates from the YARN cluster received so far. Waiting ...");
- }
+ ClusterClient client = activeCommandLine.retrieveCluster(options.getCommandLine(), config);
- try {
- Thread.sleep(500);
- }
- catch (InterruptedException e) {
- LOG.error("Interrupted while waiting for TaskManagers");
- System.err.println("Thread is interrupted");
- Thread.currentThread().interrupt();
- }
- }
- }
- else {
- if(options.getJobManagerAddress() != null) {
- jobManagerAddress = ClientUtils.parseHostPortAddress(options.getJobManagerAddress());
- writeJobManagerAddressToConfig(jobManagerAddress);
+ if (client != null) {
+ logAndSysout("Cluster retrieved");
+ } else {
--- End diff --
Yes, I have thought about letting the command-line decide whether a cluster can be created or retrieved (it would return an Enum based on the options supplied). I decided I would address this in a follow-up.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66791210
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java ---
@@ -211,18 +185,45 @@ public void run() {
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
isConnected = true;
+
+ logAndSysout("Waiting until all TaskManagers have connected");
--- End diff --
Since the `logAndSysout` method is defined in the `ClusterClient`, all log messages appear from being from that class, even though they are coming from the `YarnClusterClient`. Maybe it makes sense to override the method or add an argument for the logger.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
Thanks a lot for addressing my comments to quickly.
I'll try the code out soon.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66624415
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -980,110 +845,41 @@ protected ActorGateway getJobManagerGateway(CommandLineOptions options) throws E
}
/**
- * Retrieves a {@link Client} object from the given command line options and other parameters.
+ * Retrieves a {@link ClusterClient} object from the given command line options and other parameters.
*
* @param options Command line options which contain JobManager address
* @param programName Program name
- * @param userParallelism Given user parallelism
* @throws Exception
*/
- protected Client getClient(
+ protected ClusterClient getClient(
CommandLineOptions options,
- String programName,
- int userParallelism,
- boolean detachedMode)
- throws Exception {
- InetSocketAddress jobManagerAddress;
- int maxSlots = -1;
+ String programName) throws Exception {
- if (YARN_DEPLOY_JOBMANAGER.equals(options.getJobManagerAddress())) {
- logAndSysout("YARN cluster mode detected. Switching Log4j output to console");
-
- // Default yarn application name to use, if nothing is specified on the command line
- String applicationName = "Flink Application: " + programName;
-
- // user wants to run Flink in YARN cluster.
- CommandLine commandLine = options.getCommandLine();
- AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser
- .getFlinkYarnSessionCli()
- .withDefaultApplicationName(applicationName)
- .createFlinkYarnClient(commandLine);
-
- if (flinkYarnClient == null) {
- throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
- }
-
- // in case the main detached mode wasn't set, we don't wanna overwrite the one loaded
- // from yarn options.
- if (detachedMode) {
- flinkYarnClient.setDetachedMode(true);
- }
-
- // the number of slots available from YARN:
- int yarnTmSlots = flinkYarnClient.getTaskManagerSlots();
- if (yarnTmSlots == -1) {
- yarnTmSlots = 1;
- }
- maxSlots = yarnTmSlots * flinkYarnClient.getTaskManagerCount();
- if (userParallelism != -1) {
- int slotsPerTM = userParallelism / flinkYarnClient.getTaskManagerCount();
- logAndSysout("The YARN cluster has " + maxSlots + " slots available, " +
- "but the user requested a parallelism of " + userParallelism + " on YARN. " +
- "Each of the " + flinkYarnClient.getTaskManagerCount() + " TaskManagers " +
- "will get "+slotsPerTM+" slots.");
- flinkYarnClient.setTaskManagerSlots(slotsPerTM);
- }
-
- try {
- yarnCluster = flinkYarnClient.deploy();
- yarnCluster.connectToCluster();
- }
- catch (Exception e) {
- throw new RuntimeException("Error deploying the YARN cluster", e);
- }
+ // Get the custom command-line (e.g. Standalone/Yarn/Mesos)
+ CustomCommandLine<?> activeCommandLine =
+ CliFrontendParser.getActiveCustomCommandLine(options.getJobManagerAddress());
--- End diff --
Why is this method located in the Parser?
I thought the parser is only responsible for parsing the command line arguments?
Maybe it makes more sense to rename this
```
loadCustomCommandLine("org.apache.flink.client.cli.DefaultCLI");
loadCustomCommandLine("org.apache.flink.yarn.cli.FlinkYarnSessionCli", "y", "yarn");
```
to parser (loadCustomParser(), and DefaultCLIParser, YarnCliParser?)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66625805
--- Diff: flink-clients/src/main/java/org/apache/flink/client/cli/CustomCommandLine.java ---
@@ -0,0 +1,65 @@
+/*
+ * 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.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+
+
+/**
+ * Custom command-line interface to load hooks for the command-line interface.
+ */
+public interface CustomCommandLine<ClusterType extends ClusterClient> {
+
+ /**
+ * Returns a unique identifier for this custom command-line.
+ * @return An unique identifier string
+ */
+ String getIdentifier();
+
+ /**
+ * Adds custom options to the existing run options.
+ * @param baseOptions The existing options.
+ */
+ void addRunOptions(Options baseOptions);
+
+ /**
+ * Adds custom options to the existing general options.
+ * @param baseOptions The existing options.
+ */
+ 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
+ * @return Client if a cluster could be retrieve, null otherwise
--- End diff --
typo: retrieved
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
While trying out your code, I noticed that the `.yarn-properties` file is not properly deleted, even though my previous yarn session was shutting down correctly:
```
2016-06-13 07:16:51,885 INFO org.apache.flink.client.program.ClusterClient - TaskManager status (0/1)
TaskManager status (0/1)
2016-06-13 07:16:52,386 INFO org.apache.flink.client.program.ClusterClient - TaskManager status (0/1)
TaskManager status (0/1)
2016-06-13 07:16:52,887 INFO org.apache.flink.client.program.ClusterClient - All TaskManagers are connected
All TaskManagers are connected
2016-06-13 07:16:52,887 INFO org.apache.flink.client.program.ClusterClient - Looking up JobManager
2016-06-13 07:16:52,912 INFO org.apache.flink.client.program.ClusterClient - Looking up JobManager
Flink JobManager is now running on 10.0.2.15:51747
JobManager Web Interface: http://quickstart.cloudera:8088/proxy/application_1447844011707_0038/
Number of connected TaskManagers changed to 1. Slots available: 1
^[[A^C2016-06-13 07:25:35,370 INFO org.apache.flink.yarn.YarnClusterClient - Shutting down YarnClusterClient from the client shutdown hook
2016-06-13 07:25:35,372 INFO org.apache.flink.yarn.YarnClusterClient - Sending shutdown request to the Application Master
2016-06-13 07:25:35,373 INFO org.apache.flink.yarn.ApplicationClient - Sending StopCluster request to JobManager.
2016-06-13 07:25:35,429 INFO org.apache.flink.yarn.ApplicationClient - Stopped Application client.
2016-06-13 07:25:35,431 INFO org.apache.flink.yarn.ApplicationClient - Disconnect from JobManager Actor[akka.tcp://flink@10.0.2.15:51747/user/jobmanager#1733798764].
2016-06-13 07:25:35,469 INFO akka.remote.RemoteActorRefProvider$RemotingTerminator - Shutting down remote daemon.
2016-06-13 07:25:35,469 INFO akka.remote.RemoteActorRefProvider$RemotingTerminator - Remote daemon shut down; proceeding with flushing remote transports.
2016-06-13 07:25:35,622 INFO akka.remote.RemoteActorRefProvider$RemotingTerminator - Remoting shut down.
2016-06-13 07:25:35,701 INFO org.apache.flink.yarn.YarnClusterClient - Deleting files in hdfs://quickstart.cloudera:8020/user/cloudera/.flink/application_1447844011707_0038
2016-06-13 07:25:35,706 INFO org.apache.flink.yarn.YarnClusterClient - Application application_1447844011707_0038 finished with state FINISHED and final state SUCCEEDED at 1465827935399
2016-06-13 07:25:36,567 INFO org.apache.flink.yarn.YarnClusterClient - YARN Client is shutting down
(reverse-i-search)`yarn': ./bin/^Crn-session.sh -n 1
[cloudera@quickstart build-target]$ ./bin/flink run -y yarn-cluster -yd -yn 1 ./examples/batch/WordCount.jar
JAR file does not exist: -y
Use the help option (-h or --help) to get help on the command.
[cloudera@quickstart build-target]$ ./bin/flink run -m yarn-cluster -yd -yn 1 ./examples/batch/WordCount.jar
2016-06-13 07:26:11,057 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - Found YARN properties file /tmp/.yarn-properties-cloudera
2016-06-13 07:26:11,057 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - Found YARN properties file /tmp/.yarn-properties-cloudera
Found YARN properties file /tmp/.yarn-properties-cloudera
2016-06-13 07:26:11,082 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - Using JobManager address from YARN properties quickstart.cloudera/10.0.2.15:51747
2016-06-13 07:26:11,082 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - Using JobManager address from YARN properties quickstart.cloudera/10.0.2.15:51747
Using JobManager address from YARN properties quickstart.cloudera/10.0.2.15:51747
2016-06-13 07:26:11,270 INFO org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at quickstart.cloudera/10.0.2.15:8032
2016-06-13 07:26:11,795 INFO org.apache.flink.yarn.YarnClusterDescriptor - Found application 'application_1447844011707_0038' with JobManager host name 'quickstart.cloudera' and port '51747' from Yarn properties file.
2016-06-13 07:26:11,842 INFO org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at quickstart.cloudera/10.0.2.15:8032
2016-06-13 07:26:11,878 ERROR org.apache.flink.yarn.YarnClusterDescriptor - The application application_1447844011707_0038 doesn't run anymore. It has previously completed with final status: SUCCEEDED
------------------------------------------------------------
The program finished with the following exception:
java.lang.RuntimeException: The Yarn application application_1447844011707_0038 doesn't run anymore.
at org.apache.flink.yarn.AbstractYarnClusterDescriptor.retrieve(AbstractYarnClusterDescriptor.java:377)
at org.apache.flink.yarn.AbstractYarnClusterDescriptor.retrieveFromConfig(AbstractYarnClusterDescriptor.java:351)
at org.apache.flink.yarn.cli.FlinkYarnSessionCli.retrieveCluster(FlinkYarnSessionCli.java:464)
at org.apache.flink.yarn.cli.FlinkYarnSessionCli.retrieveCluster(FlinkYarnSessionCli.java:62)
at org.apache.flink.client.CliFrontend.getClient(CliFrontend.java:862)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:228)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:983)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1034)
[cloudera@quickstart build-target]$
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805492
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java ---
@@ -300,18 +309,82 @@ public static boolean allocateResource(int[] nodeManagers, int toAllocate) {
return false;
}
- @Override
public void setDetachedMode(boolean detachedMode) {
this.detached = detachedMode;
}
- @Override
- public boolean isDetached() {
+ public boolean isDetachedMode() {
return detached;
}
+
+ /**
+ * Gets a Hadoop Yarn client
+ * @return Returns a YarnClient which has to be shutdown manually
+ */
+ private static YarnClient getYarnClient(Configuration conf) {
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ return yarnClient;
+ }
+
+ /**
+ * Retrieves the Yarn application and cluster from the config
+ * @param config The config with entries to retrieve the cluster
+ * @return YarnClusterClient
+ * @deprecated This should be removed in the future
+ */
+ public YarnClusterClient retrieveFromConfig(org.apache.flink.configuration.Configuration config) throws Exception {
+ String jobManagerHost = config.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int jobManagerPort = config.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+
+ if (jobManagerHost != null && jobManagerPort != -1) {
+
+ YarnClient yarnClient = getYarnClient(conf);
+ List<ApplicationReport> applicationReports = yarnClient.getApplications();
+ for (ApplicationReport report : applicationReports) {
+ if (report.getHost().equals(jobManagerHost) && report.getRpcPort() == jobManagerPort) {
+ LOG.info("Found application '{}' " +
+ "with JobManager host name '{}' and port '{}' from Yarn properties file.",
+ report.getApplicationId(), jobManagerHost, jobManagerPort);
+ return retrieve(report.getApplicationId().toString());
+ }
+ }
+
--- End diff --
Good idea.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66806299
--- Diff: docs/apis/cli.md ---
@@ -252,6 +256,12 @@ Action "list" lists running and scheduled programs.
configuration.
-r,--running Show only running programs and their JobIDs
-s,--scheduled Show only scheduled programs and their JobIDs
+ Additional arguments if -m yarn-cluster is set:
+ -yid <yarnApplicationId> YARN application ID of Flink YARN session to
+ connect to. Must not be set if JobManager HA
+ is used. In this case, JobManager RPC
+ location is automatically retrieved from
+ Zookeeper.
--- End diff --
Same as for `list` and `info`. I agree this verbosity is not very nice. Looking into how I can make this look nicer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66794535
--- Diff: docs/apis/cli.md ---
@@ -105,6 +105,10 @@ The command line can be used to
./bin/flink list -r
+- List running Flink jobs inside Flink YARN session:
+
+ ./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r
--- End diff --
Why is the client trying to connect even though the application has been finished already?
```
[cloudera@quickstart build-target]$ ./bin/flink list -m yarn-cluster -yid application_1447844011707_0036 -r
2016-06-13 06:51:34,581 INFO org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at quickstart.cloudera/10.0.2.15:8032
2016-06-13 06:51:35,017 ERROR org.apache.flink.yarn.YarnClusterDescriptor - The application application_1447844011707_0036 doesn't run anymore. It has previously completed with final status: SUCCEEDED
------------------------------------------------------------
The program finished with the following exception:
org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException: Could not retrieve the leader gateway
at org.apache.flink.runtime.util.LeaderRetrievalUtils.retrieveLeaderGateway(LeaderRetrievalUtils.java:127)
at org.apache.flink.client.CliFrontend.getJobManagerGateway(CliFrontend.java:844)
at org.apache.flink.client.CliFrontend.list(CliFrontend.java:378)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:986)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1034)
Caused by: java.util.concurrent.TimeoutException: Futures timed out after [10000 milliseconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at scala.concurrent.Await.result(package.scala)
at org.apache.flink.runtime.util.LeaderRetrievalUtils.retrieveLeaderGateway(LeaderRetrievalUtils.java:125)
... 4 more
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66801768
--- Diff: docs/setup/yarn_setup.md ---
@@ -143,6 +143,34 @@ Note that in this case its not possible to stop the YARN session using Flink.
Use the YARN utilities (`yarn application -kill <appId>`) to stop the YARN session.
+#### Attach to an existing Session
+
+Use the following command to start a session
+
+~~~bash
+./bin/yarn-session.sh
+~~~
+
+This command will show you the following overview:
+
+~~~bash
+Usage:
+ Required
+ -id,--applicationId <yarnAppId> YARN application Id
--- End diff --
I think this option is also not listed when running ./bin/yarn-session.sh.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805706
--- Diff: docs/setup/yarn_setup.md ---
@@ -143,6 +143,34 @@ Note that in this case its not possible to stop the YARN session using Flink.
Use the YARN utilities (`yarn application -kill <appId>`) to stop the YARN session.
+#### Attach to an existing Session
+
+Use the following command to start a session
+
+~~~bash
+./bin/yarn-session.sh
+~~~
+
+This command will show you the following overview:
+
+~~~bash
+Usage:
+ Required
+ -id,--applicationId <yarnAppId> YARN application Id
--- End diff --
Yes, same as above. Fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2085
I quickly tried your changes.
Some notes
(1) When listing the yarn-session.sh cli options, I could not see the newly added argument
```
./bin/yarn-session.sh
2016-06-10 09:28:37,186 ERROR org.apache.flink.yarn.cli.FlinkYarnSessionCli - Missing required argument n
Usage:
Required
-n,--container <arg> Number of YARN container to allocate (=Number of Task Managers)
Optional
-D <arg> Dynamic properties
-d,--detached Start detached
-jm,--jobManagerMemory <arg> Memory for JobManager Container [in MB]
-nm,--name <arg> 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]
Error while starting the YARN Client. Please check log output!
```
(2) Starting a YARN session works. Also, connecting with a second YARN session works. However, if I'm stopping the session from the second session client, the first one does not notice this.
It would be good if all connected clients receive the messages, instead of the last one connected.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66625502
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -1275,33 +1071,16 @@ else if (new File(CONFIG_DIRECTORY_FALLBACK_2).exists()) {
return location;
}
- public static Map<String, String> getDynamicProperties(String dynamicPropertiesEncoded) {
- if (dynamicPropertiesEncoded != null && dynamicPropertiesEncoded.length() > 0) {
- Map<String, String> properties = new HashMap<>();
-
- String[] propertyLines = dynamicPropertiesEncoded.split(CliFrontend.YARN_DYNAMIC_PROPERTIES_SEPARATOR);
- for (String propLine : propertyLines) {
- if (propLine == null) {
- continue;
- }
-
- String[] kv = propLine.split("=");
- if (kv.length >= 2 && kv[0] != null && kv[1] != null && kv[0].length() > 0) {
- properties.put(kv[0], kv[1]);
- }
- }
- return properties;
- }
- else {
- return Collections.emptyMap();
- }
- }
- public static String getYarnPropertiesLocation(Configuration conf) {
- String defaultPropertiesFileLocation = System.getProperty("java.io.tmpdir");
- String currentUser = System.getProperty("user.name");
- String propertiesFileLocation = conf.getString(ConfigConstants.YARN_PROPERTIES_FILE_LOCATION, defaultPropertiesFileLocation);
-
- return propertiesFileLocation + File.separator + CliFrontend.YARN_PROPERTIES_FILE + currentUser;
+ /**
+ * Writes the given job manager address to the associated configuration object
+ *
+ * @param address Address to write to the configuration
+ * @param config The config to write to
+ */
+ public static void writeJobManagerAddressToConfig(Configuration config, InetSocketAddress address) {
--- End diff --
How about renaming this to "setJobManagerAddressInConfig()" .. write implies that something is written to a file or something.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66789723
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java ---
@@ -181,11 +198,30 @@ public boolean getPrintStatusDuringExecution() {
}
/**
- * @return -1 if unknown. The maximum number of available processing slots at the Flink cluster
- * connected to this client.
+ * Gets the current JobManager address from the Flink configuration (may change in case of a HA setup).
+ * @return The address (host and port) of the leading JobManager
*/
- public int getMaxSlots() {
- return this.maxSlots;
+ public InetSocketAddress getJobManagerAddressFromConfig() {
+ try {
+ String hostName = flinkConfig.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int port = flinkConfig.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+ return new InetSocketAddress(hostName, port);
--- End diff --
Missing indentation
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805382
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java ---
@@ -211,18 +185,45 @@ public void run() {
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
isConnected = true;
+
+ logAndSysout("Waiting until all TaskManagers have connected");
--- End diff --
Thank you. I fixed that by using a non-static variable for the logger and dynamically retrieving the class name, i.e.
```java
private final Logger LOG = LoggerFactory.getLogger(getClass());
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66765910
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -980,110 +845,41 @@ protected ActorGateway getJobManagerGateway(CommandLineOptions options) throws E
}
/**
- * Retrieves a {@link Client} object from the given command line options and other parameters.
+ * Retrieves a {@link ClusterClient} object from the given command line options and other parameters.
*
* @param options Command line options which contain JobManager address
* @param programName Program name
- * @param userParallelism Given user parallelism
* @throws Exception
*/
- protected Client getClient(
+ protected ClusterClient getClient(
CommandLineOptions options,
- String programName,
- int userParallelism,
- boolean detachedMode)
- throws Exception {
- InetSocketAddress jobManagerAddress;
- int maxSlots = -1;
+ String programName) throws Exception {
- if (YARN_DEPLOY_JOBMANAGER.equals(options.getJobManagerAddress())) {
- logAndSysout("YARN cluster mode detected. Switching Log4j output to console");
-
- // Default yarn application name to use, if nothing is specified on the command line
- String applicationName = "Flink Application: " + programName;
-
- // user wants to run Flink in YARN cluster.
- CommandLine commandLine = options.getCommandLine();
- AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser
- .getFlinkYarnSessionCli()
- .withDefaultApplicationName(applicationName)
- .createFlinkYarnClient(commandLine);
-
- if (flinkYarnClient == null) {
- throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
- }
-
- // in case the main detached mode wasn't set, we don't wanna overwrite the one loaded
- // from yarn options.
- if (detachedMode) {
- flinkYarnClient.setDetachedMode(true);
- }
-
- // the number of slots available from YARN:
- int yarnTmSlots = flinkYarnClient.getTaskManagerSlots();
- if (yarnTmSlots == -1) {
- yarnTmSlots = 1;
- }
- maxSlots = yarnTmSlots * flinkYarnClient.getTaskManagerCount();
- if (userParallelism != -1) {
- int slotsPerTM = userParallelism / flinkYarnClient.getTaskManagerCount();
- logAndSysout("The YARN cluster has " + maxSlots + " slots available, " +
- "but the user requested a parallelism of " + userParallelism + " on YARN. " +
- "Each of the " + flinkYarnClient.getTaskManagerCount() + " TaskManagers " +
- "will get "+slotsPerTM+" slots.");
- flinkYarnClient.setTaskManagerSlots(slotsPerTM);
- }
-
- try {
- yarnCluster = flinkYarnClient.deploy();
- yarnCluster.connectToCluster();
- }
- catch (Exception e) {
- throw new RuntimeException("Error deploying the YARN cluster", e);
- }
+ // Get the custom command-line (e.g. Standalone/Yarn/Mesos)
+ CustomCommandLine<?> activeCommandLine =
+ CliFrontendParser.getActiveCustomCommandLine(options.getJobManagerAddress());
--- End diff --
Okay, moving the logic is addressing my concerns.
I checked the classes again and my renaming suggestions don't make much sense ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2085
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805960
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java ---
@@ -181,11 +198,30 @@ public boolean getPrintStatusDuringExecution() {
}
/**
- * @return -1 if unknown. The maximum number of available processing slots at the Flink cluster
- * connected to this client.
+ * Gets the current JobManager address from the Flink configuration (may change in case of a HA setup).
+ * @return The address (host and port) of the leading JobManager
*/
- public int getMaxSlots() {
- return this.maxSlots;
+ public InetSocketAddress getJobManagerAddressFromConfig() {
+ try {
+ String hostName = flinkConfig.getString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, null);
+ int port = flinkConfig.getInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, -1);
+ return new InetSocketAddress(hostName, port);
--- End diff --
Fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66625318
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -980,110 +845,41 @@ protected ActorGateway getJobManagerGateway(CommandLineOptions options) throws E
}
/**
- * Retrieves a {@link Client} object from the given command line options and other parameters.
+ * Retrieves a {@link ClusterClient} object from the given command line options and other parameters.
*
* @param options Command line options which contain JobManager address
* @param programName Program name
- * @param userParallelism Given user parallelism
* @throws Exception
*/
- protected Client getClient(
+ protected ClusterClient getClient(
CommandLineOptions options,
- String programName,
- int userParallelism,
- boolean detachedMode)
- throws Exception {
- InetSocketAddress jobManagerAddress;
- int maxSlots = -1;
+ String programName) throws Exception {
- if (YARN_DEPLOY_JOBMANAGER.equals(options.getJobManagerAddress())) {
- logAndSysout("YARN cluster mode detected. Switching Log4j output to console");
-
- // Default yarn application name to use, if nothing is specified on the command line
- String applicationName = "Flink Application: " + programName;
-
- // user wants to run Flink in YARN cluster.
- CommandLine commandLine = options.getCommandLine();
- AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser
- .getFlinkYarnSessionCli()
- .withDefaultApplicationName(applicationName)
- .createFlinkYarnClient(commandLine);
-
- if (flinkYarnClient == null) {
- throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
- }
-
- // in case the main detached mode wasn't set, we don't wanna overwrite the one loaded
- // from yarn options.
- if (detachedMode) {
- flinkYarnClient.setDetachedMode(true);
- }
-
- // the number of slots available from YARN:
- int yarnTmSlots = flinkYarnClient.getTaskManagerSlots();
- if (yarnTmSlots == -1) {
- yarnTmSlots = 1;
- }
- maxSlots = yarnTmSlots * flinkYarnClient.getTaskManagerCount();
- if (userParallelism != -1) {
- int slotsPerTM = userParallelism / flinkYarnClient.getTaskManagerCount();
- logAndSysout("The YARN cluster has " + maxSlots + " slots available, " +
- "but the user requested a parallelism of " + userParallelism + " on YARN. " +
- "Each of the " + flinkYarnClient.getTaskManagerCount() + " TaskManagers " +
- "will get "+slotsPerTM+" slots.");
- flinkYarnClient.setTaskManagerSlots(slotsPerTM);
- }
-
- try {
- yarnCluster = flinkYarnClient.deploy();
- yarnCluster.connectToCluster();
- }
- catch (Exception e) {
- throw new RuntimeException("Error deploying the YARN cluster", e);
- }
+ // Get the custom command-line (e.g. Standalone/Yarn/Mesos)
+ CustomCommandLine<?> activeCommandLine =
+ CliFrontendParser.getActiveCustomCommandLine(options.getJobManagerAddress());
- jobManagerAddress = yarnCluster.getJobManagerAddress();
- writeJobManagerAddressToConfig(jobManagerAddress);
-
- // overwrite the yarn client config (because the client parses the dynamic properties)
- this.config.addAll(flinkYarnClient.getFlinkConfiguration());
-
- logAndSysout("YARN cluster started");
- logAndSysout("JobManager web interface address " + yarnCluster.getWebInterfaceURL());
- logAndSysout("Waiting until all TaskManagers have connected");
-
- while(true) {
- GetClusterStatusResponse status = yarnCluster.getClusterStatus();
- if (status != null) {
- if (status.numRegisteredTaskManagers() < flinkYarnClient.getTaskManagerCount()) {
- logAndSysout("TaskManager status (" + status.numRegisteredTaskManagers() + "/"
- + flinkYarnClient.getTaskManagerCount() + ")");
- } else {
- logAndSysout("All TaskManagers are connected");
- break;
- }
- } else {
- logAndSysout("No status updates from the YARN cluster received so far. Waiting ...");
- }
+ ClusterClient client = activeCommandLine.retrieveCluster(options.getCommandLine(), config);
- try {
- Thread.sleep(500);
- }
- catch (InterruptedException e) {
- LOG.error("Interrupted while waiting for TaskManagers");
- System.err.println("Thread is interrupted");
- Thread.currentThread().interrupt();
- }
- }
- }
- else {
- if(options.getJobManagerAddress() != null) {
- jobManagerAddress = ClientUtils.parseHostPortAddress(options.getJobManagerAddress());
- writeJobManagerAddressToConfig(jobManagerAddress);
+ if (client != null) {
+ logAndSysout("Cluster retrieved");
+ } else {
--- End diff --
I wonder if it would be a bit more obvious to have some methods on the `ClusterClient` to check if they can retrieve a cluster or if they need to create one.
Right now its a bit implicit through the `null`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66628410
--- Diff: flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java ---
@@ -980,110 +845,41 @@ protected ActorGateway getJobManagerGateway(CommandLineOptions options) throws E
}
/**
- * Retrieves a {@link Client} object from the given command line options and other parameters.
+ * Retrieves a {@link ClusterClient} object from the given command line options and other parameters.
*
* @param options Command line options which contain JobManager address
* @param programName Program name
- * @param userParallelism Given user parallelism
* @throws Exception
*/
- protected Client getClient(
+ protected ClusterClient getClient(
CommandLineOptions options,
- String programName,
- int userParallelism,
- boolean detachedMode)
- throws Exception {
- InetSocketAddress jobManagerAddress;
- int maxSlots = -1;
+ String programName) throws Exception {
- if (YARN_DEPLOY_JOBMANAGER.equals(options.getJobManagerAddress())) {
- logAndSysout("YARN cluster mode detected. Switching Log4j output to console");
-
- // Default yarn application name to use, if nothing is specified on the command line
- String applicationName = "Flink Application: " + programName;
-
- // user wants to run Flink in YARN cluster.
- CommandLine commandLine = options.getCommandLine();
- AbstractFlinkYarnClient flinkYarnClient = CliFrontendParser
- .getFlinkYarnSessionCli()
- .withDefaultApplicationName(applicationName)
- .createFlinkYarnClient(commandLine);
-
- if (flinkYarnClient == null) {
- throw new RuntimeException("Unable to create Flink YARN Client. Check previous log messages");
- }
-
- // in case the main detached mode wasn't set, we don't wanna overwrite the one loaded
- // from yarn options.
- if (detachedMode) {
- flinkYarnClient.setDetachedMode(true);
- }
-
- // the number of slots available from YARN:
- int yarnTmSlots = flinkYarnClient.getTaskManagerSlots();
- if (yarnTmSlots == -1) {
- yarnTmSlots = 1;
- }
- maxSlots = yarnTmSlots * flinkYarnClient.getTaskManagerCount();
- if (userParallelism != -1) {
- int slotsPerTM = userParallelism / flinkYarnClient.getTaskManagerCount();
- logAndSysout("The YARN cluster has " + maxSlots + " slots available, " +
- "but the user requested a parallelism of " + userParallelism + " on YARN. " +
- "Each of the " + flinkYarnClient.getTaskManagerCount() + " TaskManagers " +
- "will get "+slotsPerTM+" slots.");
- flinkYarnClient.setTaskManagerSlots(slotsPerTM);
- }
-
- try {
- yarnCluster = flinkYarnClient.deploy();
- yarnCluster.connectToCluster();
- }
- catch (Exception e) {
- throw new RuntimeException("Error deploying the YARN cluster", e);
- }
+ // Get the custom command-line (e.g. Standalone/Yarn/Mesos)
+ CustomCommandLine<?> activeCommandLine =
+ CliFrontendParser.getActiveCustomCommandLine(options.getJobManagerAddress());
--- End diff --
Well spotted. I think I'll move this logic to the implementation of the CustomCommandLine.
I don't quite understand your renaming suggestion, are you suggesting to break up the CustomCommandLine into CustomParser and CustomCLI?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2085: [FLINK-3937] programmatic resuming of clusters
Posted by mxm <gi...@git.apache.org>.
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2085#discussion_r66805455
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java ---
@@ -415,62 +513,83 @@ public int run(String[] args) {
}
System.out.println(description);
return 0;
+ } else if (cmd.hasOption(APPLICATION_ID.getOpt())) { // TODO RM
--- End diff --
Sorry, added this TODO while I fixing some of your first comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---