Merge -c r1470797 from trunk to branch-2 for YARN-549. YarnClient.submitApplication should wait for application to be accepted by the RM (Zhijie Shen via bikas)

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

hadoop-yarn-project/CHANGES.txt

@@ -209,6 +209,9 @@ Release 2.0.5-beta - UNRELEASED
 
     YARN-594. Update test and add comments in YARN-534 (Jian He via bikas)
 
+    YARN-549. YarnClient.submitApplication should wait for application to be
+    accepted by the RM (Zhijie Shen via bikas)
+
 Release 2.0.4-alpha - UNRELEASED
 
   INCOMPATIBLE CHANGES

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

@@ -96,7 +96,9 @@ public interface ClientRMProtocol {
    * 
    * <p>Currently the <code>ResourceManager</code> sends an immediate (empty) 
    * {@link SubmitApplicationResponse} on accepting the submission and throws 
-   * an exception if it rejects the submission.</p>
+   * an exception if it rejects the submission. However, this call needs to be
+   * followed by {@link #getApplicationReport(GetApplicationReportRequest)}
+   * to make sure that the application gets properly submitted.</p>
    * 
    * <p> In secure mode,the <code>ResourceManager</code> verifies access to
    * queues etc. before accepting the application submission.</p>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/YarnClient.java

@@ -63,7 +63,9 @@ public interface YarnClient extends Service {
 
   /**
    * <p>
-   * Submit a new application to <code>YARN.</code>
+   * Submit a new application to <code>YARN.</code> It is a blocking call, such
+   * that it will not return {@link ApplicationId} until the submitted
+   * application has been submitted and accepted by the ResourceManager.
    * </p>
    * 
    * @param appContext

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

@@ -53,9 +53,11 @@ import org.apache.hadoop.yarn.api.records.DelegationToken;
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
 import org.apache.hadoop.yarn.service.AbstractService;
 import org.apache.hadoop.yarn.util.Records;
@@ -68,6 +70,7 @@ public class YarnClientImpl extends AbstractService implements YarnClient {
 
   protected ClientRMProtocol rmClient;
   protected InetSocketAddress rmAddress;
+  protected long statePollIntervalMillis;
 
   private static final String ROOT = "root";
 
@@ -90,6 +93,9 @@ public class YarnClientImpl extends AbstractService implements YarnClient {
     if (this.rmAddress == null) {
       this.rmAddress = getRmAddress(conf);
     }
+    statePollIntervalMillis = conf.getLong(
+        YarnConfiguration.YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS,
+        YarnConfiguration.DEFAULT_YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS);
     super.init(conf);
   }
 
@@ -131,6 +137,29 @@ public class YarnClientImpl extends AbstractService implements YarnClient {
         Records.newRecord(SubmitApplicationRequest.class);
     request.setApplicationSubmissionContext(appContext);
     rmClient.submitApplication(request);
+
+    int pollCount = 0;
+    while (true) {
+      YarnApplicationState state =
+          getApplicationReport(applicationId).getYarnApplicationState();
+      if (!state.equals(YarnApplicationState.NEW) &&
+          !state.equals(YarnApplicationState.NEW_SAVING)) {
+        break;
+      }
+      // Notify the client through the log every 10 poll, in case the client
+      // is blocked here too long.
+      if (++pollCount % 10 == 0) {
+        LOG.info("Application submission is not finished, " +
+            "submitted application " + applicationId +
+            " is still in " + state);
+      }
+      try {
+        Thread.sleep(statePollIntervalMillis);
+      } catch (InterruptedException ie) {
+      }
+    }
+
+
     LOG.info("Submitted application " + applicationId + " to ResourceManager"
         + " at " + rmAddress);
     return applicationId;

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

@@ -18,10 +18,25 @@
 
 package org.apache.hadoop.yarn.client;
 
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import junit.framework.Assert;
+
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.client.YarnClient;
+import org.apache.hadoop.yarn.api.ClientRMProtocol;
-import org.apache.hadoop.yarn.client.YarnClientImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.util.Records;
 import org.junit.Test;
 
 public class TestYarnClient {
@@ -43,4 +58,76 @@ public class TestYarnClient {
     client.start();
     client.stop();
   }
+
+  @Test (timeout = 30000)
+  public void testSubmitApplication() {
+    Configuration conf = new Configuration();
+    conf.setLong(YarnConfiguration.YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS,
+        100); // speed up tests
+    final YarnClient client = new MockYarnClient();
+    client.init(conf);
+    client.start();
+
+    YarnApplicationState[] exitStates = new YarnApplicationState[]
+        {
+          YarnApplicationState.SUBMITTED,
+          YarnApplicationState.ACCEPTED,
+          YarnApplicationState.RUNNING,
+          YarnApplicationState.FINISHED,
+          YarnApplicationState.FAILED,
+          YarnApplicationState.KILLED
+        };
+    for (int i = 0; i < exitStates.length; ++i) {
+      ApplicationSubmissionContext context =
+          mock(ApplicationSubmissionContext.class);
+      ApplicationId applicationId = Records.newRecord(ApplicationId.class);
+      applicationId.setClusterTimestamp(System.currentTimeMillis());
+      applicationId.setId(i);
+      when(context.getApplicationId()).thenReturn(applicationId);
+      ((MockYarnClient) client).setYarnApplicationState(exitStates[i]);
+      try {
+        client.submitApplication(context);
+      } catch (YarnRemoteException e) {
+        Assert.fail("Exception is not expected.");
+      }
+      verify(((MockYarnClient) client).mockReport,times(4 * i + 4))
+          .getYarnApplicationState();
+    }
+
+    client.stop();
+  }
+
+  private static class MockYarnClient extends YarnClientImpl {
+    private ApplicationReport mockReport;
+
+    public MockYarnClient() {
+      super();
+    }
+
+    @Override
+    public void start() {
+      rmClient = mock(ClientRMProtocol.class);
+      GetApplicationReportResponse mockResponse =
+          mock(GetApplicationReportResponse.class);
+      mockReport = mock(ApplicationReport.class);
+      try{
+        when(rmClient.getApplicationReport(any(
+            GetApplicationReportRequest.class))).thenReturn(mockResponse);
+      } catch (YarnRemoteException e) {
+        Assert.fail("Exception is not expected.");
+      }
+      when(mockResponse.getApplicationReport()).thenReturn(mockReport);
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    public void setYarnApplicationState(YarnApplicationState state) {
+      when(mockReport.getYarnApplicationState()).thenReturn(
+          YarnApplicationState.NEW, YarnApplicationState.NEW_SAVING,
+          YarnApplicationState.NEW_SAVING, state);
+    }
+  }
+
 }

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

@@ -692,6 +692,19 @@ public class YarnConfiguration extends Configuration {
    */
   public static boolean DEFAULT_YARN_MINICLUSTER_FIXED_PORTS = false;
 
+  ////////////////////////////////
+  // Other Configs
+  ////////////////////////////////
+
+  /**
+   * The interval of the yarn client's querying application state after
+   * application submission. The unit is millisecond.
+   */
+  public static final String YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS =
+      YARN_PREFIX + "client.app-submission.poll-interval";
+  public static final long DEFAULT_YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS =
+      1000;
+
   public YarnConfiguration() {
     super();
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -708,4 +708,12 @@
      <value>$HADOOP_CONF_DIR,$HADOOP_COMMON_HOME/share/hadoop/common/*,$HADOOP_COMMON_HOME/share/hadoop/common/lib/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*,$HADOOP_YARN_HOME/share/hadoop/yarn/*,$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*</value>
   </property>
 
+  <!-- Other configuration -->
+  <property>
+    <description>The interval of the yarn client's querying application state
+    after application submission. The unit is millisecond.</description>
+    <name>yarn.client.app-submission.poll-interval</name>
+    <value>1000</value>
+  </property>
+
 </configuration>