MAPREDUCE-3250. When AM restarts, client keeps reconnecting to the new AM and prints a lots of logs. (vinodkv via mahadev) - Merging r1189023 from trunk.

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

hadoop-mapreduce-project/CHANGES.txt

@@ -1715,6 +1715,9 @@ Release 0.23.0 - Unreleased
     MAPREDUCE-3269. Fixed log4j properties to correctly set logging options
     for JobHistoryServer vis-a-vis JobSummary logs. (mahadev via acmurthy) 
 
+    MAPREDUCE-3250. When AM restarts, client keeps reconnecting to the new AM 
+    and prints a lots of logs. (vinodkv via mahadev)
+
 Release 0.22.0 - Unreleased
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ClientServiceDelegate.java

@@ -81,7 +81,6 @@ public class ClientServiceDelegate {
   private final ApplicationId appId;
   private final ResourceMgrDelegate rm;
   private final MRClientProtocol historyServerProxy;
-  private boolean forceRefresh;
   private MRClientProtocol realProxy = null;
   private RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
   private static String UNKNOWN_USER = "Unknown User";
@@ -122,7 +121,7 @@ public class ClientServiceDelegate {
   }
 
   private MRClientProtocol getProxy() throws YarnRemoteException {
-    if (!forceRefresh && realProxy != null) {
+    if (realProxy != null) {
       return realProxy;
     }
     
@@ -133,7 +132,9 @@ public class ClientServiceDelegate {
       trackingUrl = application.getTrackingUrl();
     }
     String serviceAddr = null;
-    while (application == null || YarnApplicationState.RUNNING == application.getYarnApplicationState()) {
+    while (application == null
+        || YarnApplicationState.RUNNING == application
+            .getYarnApplicationState()) {
       if (application == null) {
         LOG.info("Could not get Job info from RM for job " + jobId
             + ". Redirecting to job history server.");
@@ -163,7 +164,7 @@ public class ClientServiceDelegate {
         }
         LOG.info("Tracking Url of JOB is " + application.getTrackingUrl());
         LOG.info("Connecting to " + serviceAddr);
-        instantiateAMProxy(serviceAddr);
+        realProxy = instantiateAMProxy(serviceAddr);
         return realProxy;
       } catch (IOException e) {
         //possibly the AM has crashed
@@ -233,10 +234,12 @@ public class ClientServiceDelegate {
     return historyServerProxy;
   }
 
-  private void instantiateAMProxy(final String serviceAddr) throws IOException {
+  MRClientProtocol instantiateAMProxy(final String serviceAddr)
+      throws IOException {
     UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
     LOG.trace("Connecting to ApplicationMaster at: " + serviceAddr);
-    realProxy = currentUser.doAs(new PrivilegedAction<MRClientProtocol>() {
+    MRClientProtocol proxy = currentUser
+        .doAs(new PrivilegedAction<MRClientProtocol>() {
       @Override
       public MRClientProtocol run() {
         YarnRPC rpc = YarnRPC.create(conf);
@@ -245,6 +248,7 @@ public class ClientServiceDelegate {
       }
     });
     LOG.trace("Connected to ApplicationMaster at: " + serviceAddr);
+    return proxy;
   }
 
   private synchronized Object invoke(String method, Class argClass,
@@ -274,12 +278,14 @@ public class ClientServiceDelegate {
             " retrying..");
         LOG.debug("Failed exception on AM/History contact", 
             e.getTargetException());
-        forceRefresh = true;
+        // Force reconnection by setting the proxy to null.
+        realProxy = null;
       } catch (Exception e) {
         LOG.info("Failed to contact AM/History for job " + jobId
             + "  Will retry..");
         LOG.debug("Failing to contact application master", e);
-        forceRefresh = true;
+        // Force reconnection by setting the proxy to null.
+        realProxy = null;
       }
     }
   }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java

@@ -68,8 +68,6 @@ import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.yarn.YarnException;
 import org.apache.hadoop.yarn.api.ClientRMProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
@@ -84,6 +82,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -123,20 +123,24 @@ public class TestClientRedirect {
     conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_FRAMEWORK_NAME);
     conf.set(YarnConfiguration.RM_ADDRESS, RMADDRESS);
     conf.set(JHAdminConfig.MR_HISTORY_ADDRESS, HSHOSTADDRESS);
+
+    // Start the RM.
     RMService rmService = new RMService("test");
     rmService.init(conf);
     rmService.start();
 
+    // Start the AM.
     AMService amService = new AMService();
     amService.init(conf);
     amService.start(conf);
-    amRunning = true;
 
+    // Start the HS.
     HistoryService historyService = new HistoryService();
     historyService.init(conf);
     historyService.start(conf);
 
     LOG.info("services started");
+
     Cluster cluster = new Cluster(conf);
     org.apache.hadoop.mapreduce.JobID jobID =
       new org.apache.hadoop.mapred.JobID("201103121733", 1);
@@ -151,13 +155,13 @@ public class TestClientRedirect {
 
     //bring down the AM service
     amService.stop();
-    amRunning = false;
 
     LOG.info("Sleeping for 5 seconds after stop for" +
     		" the server to exit cleanly..");
     Thread.sleep(5000);
 
     amRestarting = true;
+
     // Same client
     //results are returned from fake (not started job)
     counters = cluster.getJob(jobID).getCounters();
@@ -181,14 +185,15 @@ public class TestClientRedirect {
     amService = new AMService();
     amService.init(conf);
     amService.start(conf);
-    amRunning = true;
     amContact = false; //reset
 
     counters = cluster.getJob(jobID).getCounters();
     validateCounters(counters);
     Assert.assertTrue(amContact);
 
-    amRunning = false;
+    // Stop the AM. It is not even restarting. So it should be treated as
+    // completed.
+    amService.stop();
 
     // Same client
     counters = cluster.getJob(jobID).getCounters();
@@ -347,6 +352,7 @@ public class TestClientRedirect {
     private InetSocketAddress bindAddress;
     private Server server;
     private final String hostAddress;
+
     public AMService() {
       this(AMHOSTADDRESS);
     }
@@ -376,11 +382,13 @@ public class TestClientRedirect {
         NetUtils.createSocketAddr(hostNameResolved.getHostAddress()
             + ":" + server.getPort());
        super.start();
+       amRunning = true;
     }
 
     public void stop() {
       server.stop();
       super.stop();
+      amRunning = false;
     }
 
     @Override

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientServiceDelegate.java

@@ -18,8 +18,11 @@
 
 package org.apache.hadoop.mapred;
 
-import static org.mockito.Mockito.mock;
+import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
+
+import java.io.IOException;
+
 import junit.framework.Assert;
 
 import org.apache.hadoop.conf.Configuration;
@@ -32,12 +35,14 @@ import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetJobReportRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetJobReportResponse;
 import org.apache.hadoop.mapreduce.v2.api.records.JobReport;
 import org.apache.hadoop.mapreduce.v2.api.records.JobState;
-import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 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.ipc.RPCUtil;
+import org.apache.hadoop.yarn.util.BuilderUtils;
 import org.apache.hadoop.yarn.util.Records;
 import org.junit.Test;
 
@@ -102,6 +107,8 @@ public class TestClientServiceDelegate {
 
     JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
     Assert.assertNotNull(jobStatus);
+    verify(historyServerProxy, times(4)).getJobReport(
+        any(GetJobReportRequest.class));
   }
 
   @Test
@@ -115,7 +122,7 @@ public class TestClientServiceDelegate {
 
     //RM has app report and job History Server is not configured
     ResourceMgrDelegate rm = mock(ResourceMgrDelegate.class);
-    ApplicationReport applicationReport = getApplicationReport();
+    ApplicationReport applicationReport = getFinishedApplicationReport();
     when(rm.getApplicationReport(jobId.getAppId())).thenReturn(
         applicationReport);
 
@@ -125,7 +132,6 @@ public class TestClientServiceDelegate {
     Assert.assertEquals(JobStatus.State.SUCCEEDED, jobStatus.getState());
   }
   
-  
   @Test
   public void testJobReportFromHistoryServer() throws Exception {                                 
     MRClientProtocol historyServerProxy = mock(MRClientProtocol.class);                           
@@ -145,6 +151,66 @@ public class TestClientServiceDelegate {
     Assert.assertEquals(1.0f, jobStatus.getReduceProgress());                                     
   }
 
+  @Test
+  public void testReconnectOnAMRestart() throws IOException {
+
+    MRClientProtocol historyServerProxy = mock(MRClientProtocol.class);
+
+    // RM returns AM1 url, null, null and AM2 url on invocations.
+    // Nulls simulate the time when AM2 is in the process of restarting.
+    ResourceMgrDelegate rmDelegate = mock(ResourceMgrDelegate.class);
+    when(rmDelegate.getApplicationReport(jobId.getAppId())).thenReturn(
+        getRunningApplicationReport("am1", 78)).thenReturn(
+        getRunningApplicationReport(null, 0)).thenReturn(
+        getRunningApplicationReport(null, 0)).thenReturn(
+        getRunningApplicationReport("am2", 90));
+
+    GetJobReportResponse jobReportResponse1 = mock(GetJobReportResponse.class);
+    when(jobReportResponse1.getJobReport()).thenReturn(
+        MRBuilderUtils.newJobReport(jobId, "jobName-firstGen", "user",
+            JobState.RUNNING, 0, 0, 0, 0, 0, 0, "anything"));
+
+    // First AM returns a report with jobName firstGen and simulates AM shutdown
+    // on second invocation.
+    MRClientProtocol firstGenAMProxy = mock(MRClientProtocol.class);
+    when(firstGenAMProxy.getJobReport(any(GetJobReportRequest.class)))
+        .thenReturn(jobReportResponse1).thenThrow(
+            new RuntimeException("AM is down!"));
+
+    GetJobReportResponse jobReportResponse2 = mock(GetJobReportResponse.class);
+    when(jobReportResponse2.getJobReport()).thenReturn(
+        MRBuilderUtils.newJobReport(jobId, "jobName-secondGen", "user",
+            JobState.RUNNING, 0, 0, 0, 0, 0, 0, "anything"));
+
+    // Second AM generation returns a report with jobName secondGen
+    MRClientProtocol secondGenAMProxy = mock(MRClientProtocol.class);
+    when(secondGenAMProxy.getJobReport(any(GetJobReportRequest.class)))
+        .thenReturn(jobReportResponse2);
+
+    ClientServiceDelegate clientServiceDelegate = spy(getClientServiceDelegate(
+        historyServerProxy, rmDelegate));
+    // First time, connection should be to AM1, then to AM2. Further requests
+    // should use the same proxy to AM2 and so instantiateProxy shouldn't be
+    // called.
+    doReturn(firstGenAMProxy).doReturn(secondGenAMProxy).when(
+        clientServiceDelegate).instantiateAMProxy(any(String.class));
+
+    JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("jobName-firstGen", jobStatus.getJobName());
+
+    jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("jobName-secondGen", jobStatus.getJobName());
+
+    jobStatus = clientServiceDelegate.getJobStatus(oldJobId);
+    Assert.assertNotNull(jobStatus);
+    Assert.assertEquals("jobName-secondGen", jobStatus.getJobName());
+
+    verify(clientServiceDelegate, times(2)).instantiateAMProxy(
+        any(String.class));
+  }
+
   private GetJobReportRequest getJobReportRequest() {
     GetJobReportRequest request = Records.newRecord(GetJobReportRequest.class);
     request.setJobId(jobId);
@@ -161,20 +227,18 @@ public class TestClientServiceDelegate {
     return jobReportResponse;
   }
 
-  private ApplicationReport getApplicationReport() {
+  private ApplicationReport getFinishedApplicationReport() {
-    ApplicationReport applicationReport = Records
+    return BuilderUtils.newApplicationReport(BuilderUtils.newApplicationId(
-        .newRecord(ApplicationReport.class);
+        1234, 5), "user", "queue", "appname", "host", 124, null,
-    applicationReport.setYarnApplicationState(YarnApplicationState.FINISHED);
+        YarnApplicationState.FINISHED, "diagnostics", "url", 0, 0,
-    applicationReport.setUser("root");
+        FinalApplicationStatus.SUCCEEDED, null);
-    applicationReport.setHost("N/A");
+  }
-    applicationReport.setName("N/A");
+
-    applicationReport.setQueue("N/A");
+  private ApplicationReport getRunningApplicationReport(String host, int port) {
-    applicationReport.setStartTime(0);
+    return BuilderUtils.newApplicationReport(BuilderUtils.newApplicationId(
-    applicationReport.setFinishTime(0);
+        1234, 5), "user", "queue", "appname", host, port, null,
-    applicationReport.setTrackingUrl("N/A");
+        YarnApplicationState.RUNNING, "diagnostics", "url", 0, 0,
-    applicationReport.setDiagnostics("N/A");
+        FinalApplicationStatus.UNDEFINED, null);
-    applicationReport.setFinalApplicationStatus(FinalApplicationStatus.SUCCEEDED);
-    return applicationReport;
   }
 
   private ResourceMgrDelegate getRMDelegate() throws YarnRemoteException {