YARN-1630. Introduce timeout for async polling operations in YarnClientImpl (Aditya Acharya via Sandy Ryza)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2.3@1562294 13f79535-47bb-0310-9956-ffa450edef68

hadoop-yarn-project/CHANGES.txt

@@ -219,6 +219,9 @@ Release 2.4.0 - UNRELEASED
     YARN-1573. ZK store should use a private password for root-node-acls. 
     (kasha).
 
+    YARN-1630. Introduce timeout for async polling operations in YarnClientImpl
+    (Aditya Acharya via Sandy Ryza)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -952,6 +952,17 @@ public class YarnConfiguration extends Configuration {
       YARN_PREFIX + "client.application-client-protocol.poll-interval-ms";
   public static final long DEFAULT_YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_INTERVAL_MS =
       200;
+
+  /**
+   * The duration that the yarn client library waits, cumulatively across polls,
+   * for an expected state change to occur. Defaults to -1, which indicates no
+   * limit.
+   */
+  public static final String YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_TIMEOUT_MS =
+      YARN_PREFIX + "client.application-client-protocol.poll-timeout-ms";
+  public static final long DEFAULT_YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_TIMEOUT_MS =
+      -1;
+
   /**
    * Max number of threads in NMClientAsync to process container management
    * events

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java

@@ -80,6 +80,7 @@ public class YarnClientImpl extends YarnClient {
   protected ApplicationClientProtocol rmClient;
   protected long submitPollIntervalMillis;
   private long asyncApiPollIntervalMillis;
+  private long asyncApiPollTimeoutMillis;
 
   private static final String ROOT = "root";
 
@@ -93,6 +94,9 @@ public class YarnClientImpl extends YarnClient {
     asyncApiPollIntervalMillis =
         conf.getLong(YarnConfiguration.YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_INTERVAL_MS,
           YarnConfiguration.DEFAULT_YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_INTERVAL_MS);
+    asyncApiPollTimeoutMillis =
+        conf.getLong(YarnConfiguration.YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_TIMEOUT_MS,
+            YarnConfiguration.DEFAULT_YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_TIMEOUT_MS);
     submitPollIntervalMillis = asyncApiPollIntervalMillis;
     if (conf.get(YarnConfiguration.YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS)
         != null) {
@@ -152,13 +156,24 @@ public class YarnClientImpl extends YarnClient {
     rmClient.submitApplication(request);
 
     int pollCount = 0;
+    long startTime = System.currentTimeMillis();
+
     while (true) {
       YarnApplicationState state =
           getApplicationReport(applicationId).getYarnApplicationState();
       if (!state.equals(YarnApplicationState.NEW) &&
           !state.equals(YarnApplicationState.NEW_SAVING)) {
+        LOG.info("Submitted application " + applicationId);
         break;
       }
+
+      long elapsedMillis = System.currentTimeMillis() - startTime;
+      if (enforceAsyncAPITimeout() &&
+          elapsedMillis >= asyncApiPollTimeoutMillis) {
+        throw new YarnException("Timed out while waiting for application " +
+          applicationId + " to be submitted successfully");
+      }
+
       // Notify the client through the log every 10 poll, in case the client
       // is blocked here too long.
       if (++pollCount % 10 == 0) {
@@ -169,10 +184,11 @@ public class YarnClientImpl extends YarnClient {
       try {
         Thread.sleep(submitPollIntervalMillis);
       } catch (InterruptedException ie) {
+        LOG.error("Interrupted while waiting for application " + applicationId
+            + " to be successfully submitted.");
       }
     }
 
-    LOG.info("Submitted application " + applicationId);
     return applicationId;
   }
 
@@ -185,15 +201,25 @@ public class YarnClientImpl extends YarnClient {
 
     try {
       int pollCount = 0;
+      long startTime = System.currentTimeMillis();
+
       while (true) {
         KillApplicationResponse response =
             rmClient.forceKillApplication(request);
         if (response.getIsKillCompleted()) {
+          LOG.info("Killed application " + applicationId);
           break;
         }
+
+        long elapsedMillis = System.currentTimeMillis() - startTime;
+        if (enforceAsyncAPITimeout() &&
+            elapsedMillis >= this.asyncApiPollTimeoutMillis) {
+          throw new YarnException("Timed out while waiting for application " +
+            applicationId + " to be killed.");
+        }
+
         if (++pollCount % 10 == 0) {
-          LOG.info("Watiting for application " + applicationId
-              + " to be killed.");
+          LOG.info("Waiting for application " + applicationId + " to be killed.");
         }
         Thread.sleep(asyncApiPollIntervalMillis);
       }
@@ -201,7 +227,11 @@ public class YarnClientImpl extends YarnClient {
       LOG.error("Interrupted while waiting for application " + applicationId
           + " to be killed.");
     }
-    LOG.info("Killed application " + applicationId);
+  }
+
+  @VisibleForTesting
+  boolean enforceAsyncAPITimeout() {
+    return asyncApiPollTimeoutMillis >= 0;
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.client.api.impl;
 
+import static org.junit.Assert.assertFalse;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -35,6 +36,7 @@ import java.util.Set;
 
 import junit.framework.Assert;
 
+import org.apache.commons.io.IOUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
@@ -476,4 +478,30 @@ public class TestYarnClient {
     }
   }
 
+  @Test
+  public void testAsyncAPIPollTimeout() {
+    testAsyncAPIPollTimeoutHelper(null, false);
+    testAsyncAPIPollTimeoutHelper(0L, true);
+    testAsyncAPIPollTimeoutHelper(1L, true);
+  }
+
+  private void testAsyncAPIPollTimeoutHelper(Long valueForTimeout,
+      boolean expectedTimeoutEnforcement) {
+    YarnClientImpl client = new YarnClientImpl();
+    try {
+      Configuration conf = new Configuration();
+      if (valueForTimeout != null) {
+        conf.setLong(
+            YarnConfiguration.YARN_CLIENT_APPLICATION_CLIENT_PROTOCOL_POLL_TIMEOUT_MS,
+            valueForTimeout);
+      }
+
+      client.init(conf);
+
+      Assert.assertEquals(
+          expectedTimeoutEnforcement, client.enforceAsyncAPITimeout());
+    } finally {
+      IOUtils.closeQuietly(client);
+    }
+  }
 }