You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2017/05/26 08:42:02 UTC
[3/8] flink git commit: [FLINK-6708] [yarn] Harden
FlinkYarnSessionCli to handle GetClusterStatusResponse exceptions
[FLINK-6708] [yarn] Harden FlinkYarnSessionCli to handle GetClusterStatusResponse exceptions
This PR hardens the FlinkYarnSessionCli by handling exceptions which occur when
retrieving the GetClusterStatusResponse. If no such response is retrieved and instead
an exception is thrown, the Cli won't fail but retry it the next time.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9bc34bfc
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9bc34bfc
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9bc34bfc
Branch: refs/heads/release-1.3
Commit: 9bc34bfcfe8638eed89ca6063da87148d1152016
Parents: 99e15dd
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed May 24 18:26:57 2017 +0200
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Fri May 26 16:33:13 2017 +0800
----------------------------------------------------------------------
.../apache/flink/yarn/cli/FlinkYarnSessionCli.java | 16 ++++++++++------
1 file changed, 10 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9bc34bfc/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 1ece264..d2a4340 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -413,14 +413,18 @@ public class FlinkYarnSessionCli implements CustomCommandLine<YarnClusterClient>
while (true) {
// ------------------ check if there are updates by the cluster -----------
- GetClusterStatusResponse status = yarnCluster.getClusterStatus();
- LOG.debug("Received status message: {}", status);
+ try {
+ GetClusterStatusResponse status = yarnCluster.getClusterStatus();
+ LOG.debug("Received status message: {}", status);
- if (status != null && numTaskmanagers != status.numRegisteredTaskManagers()) {
- System.err.println("Number of connected TaskManagers changed to " +
+ if (status != null && numTaskmanagers != status.numRegisteredTaskManagers()) {
+ System.err.println("Number of connected TaskManagers changed to " +
status.numRegisteredTaskManagers() + ". " +
- "Slots available: " + status.totalNumberOfSlots());
- numTaskmanagers = status.numRegisteredTaskManagers();
+ "Slots available: " + status.totalNumberOfSlots());
+ numTaskmanagers = status.numRegisteredTaskManagers();
+ }
+ } catch (Exception e) {
+ LOG.warn("Could not retrieve the current cluster status. Retrying...", e);
}
List<String> messages = yarnCluster.getNewMessages();