diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java index e5bb6e5..5f58e6a 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/AdminService.java @@ -295,6 +295,7 @@ public synchronized void monitorHealth() } } + @SuppressWarnings("unchecked") @Override public synchronized void transitionToActive( HAServiceProtocol.StateChangeRequestInfo reqInfo) throws IOException { @@ -310,10 +311,6 @@ public synchronized void transitionToActive( checkHaStateChange(reqInfo); try { rm.transitionToActive(); - // call all refresh*s for active RM to get the updated configurations. - refreshAll(); - RMAuditLogger.logSuccess(user.getShortUserName(), - "transitionToActive", "RMHAProtocolService"); } catch (Exception e) { RMAuditLogger.logFailure(user.getShortUserName(), "transitionToActive", "", "RMHAProtocolService", @@ -321,6 +318,21 @@ public synchronized void transitionToActive( throw new ServiceFailedException( "Error when transitioning to Active mode", e); } + try { + // call all refresh*s for active RM to get the updated configurations. + refreshAll(); + } catch (Exception e) { + if (isRMActive() && YarnConfiguration.shouldRMFailFast(getConfig())) { + rmContext.getDispatcher().getEventHandler() + .handle(new RMFatalEvent(RMFatalEventType.ACTIVE_REFRESH_FAIL, e)); + }else{ + rm.handleTransitionToStandBy(); + throw new ServiceFailedException( + "Error on refreshAll during transistion to Active", e); + } + } + RMAuditLogger.logSuccess(user.getShortUserName(), "transitionToActive", + "RMHAProtocolService"); } @Override diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMFatalEventType.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMFatalEventType.java index 789c018..df83c6c 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMFatalEventType.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMFatalEventType.java @@ -26,5 +26,8 @@ STATE_STORE_OP_FAILED, // Source <- Embedded Elector - EMBEDDED_ELECTOR_FAILED + EMBEDDED_ELECTOR_FAILED, + + // Source <- Admin Service + ACTIVE_REFRESH_FAIL } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMHA.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMHA.java index 0200e85..9c87a05 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMHA.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMHA.java @@ -23,6 +23,8 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import io.netty.channel.MessageSizeEstimator.Handle; + import java.io.IOException; import java.net.InetSocketAddress; @@ -42,7 +44,9 @@ import org.apache.hadoop.service.AbstractService; import org.apache.hadoop.yarn.conf.HAUtil; import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.event.AsyncDispatcher; import org.apache.hadoop.yarn.event.Dispatcher; +import org.apache.hadoop.yarn.event.Event; import org.apache.hadoop.yarn.event.EventHandler; import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; @@ -52,6 +56,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; import org.codehaus.jettison.json.JSONException; import org.codehaus.jettison.json.JSONObject; import org.junit.Assert; @@ -577,6 +582,68 @@ protected void serviceStart() throws Exception { assertEquals(0, rm.getRMContext().getRMApps().size()); } + @Test(timeout = 90000) + public void testTransitionedToActiveRefreshFail() throws Exception { + configuration.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false); + YarnConfiguration conf = new YarnConfiguration(configuration); + configuration = new CapacitySchedulerConfiguration(conf); + rm = new MockRM(configuration) { + @Override + protected AdminService createAdminService() { + return new AdminService(this, getRMContext()) { + @Override + protected void setConfig(Configuration conf) { + super.setConfig(configuration); + } + }; + } + + @Override + protected Dispatcher createDispatcher() { + return new FatalEventCountDispatcher(); + } + }; + + rm.init(configuration); + rm.start(); + final StateChangeRequestInfo requestInfo = + new StateChangeRequestInfo( + HAServiceProtocol.RequestSource.REQUEST_BY_USER); + // Transition to active call with invalid value + configuration.set("yarn.scheduler.capacity.root.default.capacity", "200"); + try { + rm.adminService.transitionToActive(requestInfo); + } catch (Exception e) { + assertTrue("Error when transitioning to Active mode".contains(e + .getMessage())); + } + assertEquals(HAServiceState.ACTIVE, rm.getRMContext().getHAServiceState()); + + configuration.set("yarn.scheduler.capacity.root.default.capacity", "100"); + rm.adminService.transitionToStandby(requestInfo); + assertEquals(HAServiceState.STANDBY, rm.getRMContext().getHAServiceState()); + configuration.set("yarn.scheduler.capacity.root.default.capacity", "200"); + try { + rm.adminService.transitionToActive(requestInfo); + } catch (Exception e) { + assertTrue("Error when transitioning to Active mode".contains(e + .getMessage())); + } + assertEquals(1, ((FatalEventCountDispatcher) rm.getRMContext() + .getDispatcher()).getEventCount()); + configuration.set("yarn.resourcemanager.fail-fast", "false"); + try { + rm.adminService.transitionToActive(requestInfo); + } catch (Exception e) { + assertTrue("Error on refreshAll during transistion to Active".contains(e + .getMessage())); + } + // Making correct conf and check the state + configuration.set("yarn.scheduler.capacity.root.default.capacity", "100"); + rm.adminService.transitionToActive(requestInfo); + assertEquals(HAServiceState.ACTIVE, rm.getRMContext().getHAServiceState()); + } + public void innerTestHAWithRMHostName(boolean includeBindHost) { //this is run two times, with and without a bind host configured if (includeBindHost) { @@ -713,4 +780,21 @@ public boolean isStopped() { return this.stopped; } } + + class FatalEventCountDispatcher extends AsyncDispatcher { + int eventreceived = 0; + + @Override + protected void dispatch(Event event) { + if (event.getType() == RMFatalEventType.ACTIVE_REFRESH_FAIL) { + eventreceived++; + } else { + super.dispatch(event); + } + } + + public int getEventCount() { + return eventreceived; + } + } }