| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.yarn.client; |
| |
| import java.nio.ByteBuffer; |
| import java.util.HashMap; |
| import java.util.Map; |
| import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse; |
| import org.apache.hadoop.yarn.api.records.CollectorInfo; |
| import org.apache.hadoop.yarn.api.records.ApplicationAccessType; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| |
| import org.apache.hadoop.yarn.server.resourcemanager.HATestUtil; |
| import org.junit.Assert; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.ha.ClientBaseWithFixes; |
| import org.apache.hadoop.ha.HAServiceProtocol; |
| import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.CancelDelegationTokenResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationAttemptReportRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationAttemptReportResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationAttemptsRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationAttemptsResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainerReportRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainerReportResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainersRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainersResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetDelegationTokenResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; |
| 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.MoveApplicationAcrossQueuesRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.MoveApplicationAcrossQueuesResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.RenewDelegationTokenRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.RenewDelegationTokenResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptReport; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.ApplicationReport; |
| import org.apache.hadoop.yarn.api.records.Container; |
| import org.apache.hadoop.yarn.api.records.ContainerId; |
| import org.apache.hadoop.yarn.api.records.ContainerReport; |
| import org.apache.hadoop.yarn.api.records.ContainerState; |
| import org.apache.hadoop.yarn.api.records.ContainerStatus; |
| import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; |
| import org.apache.hadoop.yarn.api.records.NMToken; |
| import org.apache.hadoop.yarn.api.records.NodeId; |
| import org.apache.hadoop.yarn.api.records.NodeReport; |
| import org.apache.hadoop.yarn.api.records.NodeState; |
| import org.apache.hadoop.yarn.api.records.QueueACL; |
| import org.apache.hadoop.yarn.api.records.QueueInfo; |
| import org.apache.hadoop.yarn.api.records.QueueState; |
| import org.apache.hadoop.yarn.api.records.QueueUserACLInfo; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.Token; |
| import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState; |
| import org.apache.hadoop.yarn.api.records.YarnApplicationState; |
| import org.apache.hadoop.yarn.api.records.YarnClusterMetrics; |
| import org.apache.hadoop.yarn.client.api.YarnClient; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.yarn.server.MiniYARNCluster; |
| import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatRequest; |
| import org.apache.hadoop.yarn.server.api.protocolrecords.NodeHeartbeatResponse; |
| import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerRequest; |
| import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerResponse; |
| import org.apache.hadoop.yarn.server.resourcemanager.AdminService; |
| import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService; |
| import org.apache.hadoop.yarn.server.resourcemanager.ClientRMService; |
| import org.apache.hadoop.yarn.server.resourcemanager.NMLivelinessMonitor; |
| import org.apache.hadoop.yarn.server.resourcemanager.NodesListManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.RMAppManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.RMContext; |
| import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.ResourceTrackerService; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager; |
| import org.apache.hadoop.yarn.server.security.ApplicationACLsManager; |
| import org.apache.hadoop.yarn.util.Records; |
| import org.junit.After; |
| import org.junit.Before; |
| |
| |
| /** |
| * Test Base for ResourceManager's Protocol on HA. |
| * |
| * Limited scope: |
| * For all the test cases, we only test whether the method will be re-entered |
| * when failover happens. Does not cover the entire logic of test. |
| * |
| * Test strategy: |
| * Create a separate failover thread with a trigger flag, |
| * override all APIs that are added trigger flag. |
| * When the APIs are called, we will set trigger flag as true to kick off |
| * the failover. So We can make sure the failover happens during process |
| * of the method. If this API is marked as @Idempotent or @AtMostOnce, |
| * the test cases will pass; otherwise, they will throw the exception. |
| * |
| */ |
| public abstract class ProtocolHATestBase extends ClientBaseWithFixes { |
| protected static final HAServiceProtocol.StateChangeRequestInfo req = |
| new HAServiceProtocol.StateChangeRequestInfo( |
| HAServiceProtocol.RequestSource.REQUEST_BY_USER); |
| |
| protected static final String RM1_NODE_ID = "rm1"; |
| protected static final int RM1_PORT_BASE = 10000; |
| protected static final String RM2_NODE_ID = "rm2"; |
| protected static final int RM2_PORT_BASE = 20000; |
| |
| protected Configuration conf; |
| protected MiniYARNClusterForHATesting cluster; |
| |
| protected Thread failoverThread = null; |
| private volatile boolean keepRunning; |
| |
| @Before |
| public void setup() throws IOException { |
| failoverThread = null; |
| keepRunning = true; |
| conf = new YarnConfiguration(); |
| conf.setBoolean(YarnConfiguration.RM_HA_ENABLED, true); |
| conf.setInt(YarnConfiguration.CLIENT_FAILOVER_MAX_ATTEMPTS, 5); |
| conf.set(YarnConfiguration.RM_HA_IDS, RM1_NODE_ID + "," + RM2_NODE_ID); |
| HATestUtil.setRpcAddressForRM(RM1_NODE_ID, RM1_PORT_BASE, conf); |
| HATestUtil.setRpcAddressForRM(RM2_NODE_ID, RM2_PORT_BASE, conf); |
| |
| conf.setLong(YarnConfiguration.CLIENT_FAILOVER_SLEEPTIME_BASE_MS, 100L); |
| |
| conf.setBoolean(YarnConfiguration.YARN_MINICLUSTER_FIXED_PORTS, true); |
| conf.setBoolean(YarnConfiguration.YARN_MINICLUSTER_USE_RPC, true); |
| } |
| |
| @After |
| public void teardown() throws Exception { |
| keepRunning = false; |
| if (failoverThread != null) { |
| failoverThread.interrupt(); |
| try { |
| failoverThread.join(); |
| } catch (InterruptedException ex) { |
| LOG.error("Error joining with failover thread", ex); |
| } |
| } |
| cluster.stop(); |
| } |
| |
| protected AdminService getAdminService(int index) { |
| return cluster.getResourceManager(index).getRMContext() |
| .getRMAdminService(); |
| } |
| |
| protected void explicitFailover() throws IOException { |
| int activeRMIndex = cluster.getActiveRMIndex(); |
| int newActiveRMIndex = (activeRMIndex + 1) % 2; |
| getAdminService(activeRMIndex).transitionToStandby(req); |
| getAdminService(newActiveRMIndex).transitionToActive(req); |
| assertEquals("Failover failed", newActiveRMIndex, |
| cluster.getActiveRMIndex()); |
| } |
| |
| protected YarnClient createAndStartYarnClient(Configuration conf) { |
| Configuration configuration = new YarnConfiguration(conf); |
| YarnClient client = YarnClient.createYarnClient(); |
| client.init(configuration); |
| client.start(); |
| return client; |
| } |
| |
| protected void verifyConnections() throws InterruptedException, |
| YarnException { |
| assertTrue("NMs failed to connect to the RM", |
| cluster.waitForNodeManagersToConnect(5000)); |
| verifyClientConnection(); |
| } |
| |
| protected void verifyClientConnection() { |
| int numRetries = 3; |
| while(numRetries-- > 0) { |
| Configuration conf = new YarnConfiguration(this.conf); |
| YarnClient client = createAndStartYarnClient(conf); |
| try { |
| Thread.sleep(100); |
| client.getApplications(); |
| return; |
| } catch (Exception e) { |
| LOG.error(e.getMessage()); |
| } finally { |
| client.stop(); |
| } |
| } |
| fail("Client couldn't connect to the Active RM"); |
| } |
| |
| protected Thread createAndStartFailoverThread() { |
| Thread failoverThread = new Thread() { |
| public void run() { |
| keepRunning = true; |
| while (keepRunning) { |
| if (cluster.getStartFailoverFlag()) { |
| try { |
| explicitFailover(); |
| keepRunning = false; |
| cluster.resetFailoverTriggeredFlag(true); |
| } catch (Exception e) { |
| // Do Nothing |
| } finally { |
| keepRunning = false; |
| } |
| } |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| // DO NOTHING |
| } |
| } |
| } |
| }; |
| failoverThread.start(); |
| return failoverThread; |
| } |
| |
| protected void startHACluster(int numOfNMs, boolean overrideClientRMService, |
| boolean overrideRTS, boolean overrideApplicationMasterService) |
| throws Exception { |
| conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true); |
| conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false); |
| cluster = |
| new MiniYARNClusterForHATesting(TestRMFailover.class.getName(), 2, |
| numOfNMs, 1, 1, false, overrideClientRMService, overrideRTS, |
| overrideApplicationMasterService); |
| cluster.resetStartFailoverFlag(false); |
| cluster.init(conf); |
| cluster.start(); |
| assertFalse("RM never turned active", -1 == cluster.getActiveRMIndex()); |
| verifyConnections(); |
| |
| // Do the failover |
| explicitFailover(); |
| verifyConnections(); |
| |
| failoverThread = createAndStartFailoverThread(); |
| |
| } |
| |
| protected ResourceManager getActiveRM() { |
| return cluster.getResourceManager(cluster.getActiveRMIndex()); |
| } |
| |
| public class MiniYARNClusterForHATesting extends MiniYARNCluster { |
| |
| private boolean overrideClientRMService; |
| private boolean overrideRTS; |
| private boolean overrideApplicationMasterService; |
| private final AtomicBoolean startFailover = new AtomicBoolean(false); |
| private final AtomicBoolean failoverTriggered = new AtomicBoolean(false); |
| |
| public MiniYARNClusterForHATesting(String testName, |
| int numResourceManagers, int numNodeManagers, int numLocalDirs, |
| int numLogDirs, boolean enableAHS, boolean overrideClientRMService, |
| boolean overrideRTS, boolean overrideApplicationMasterService) { |
| super(testName, numResourceManagers, numNodeManagers, numLocalDirs, |
| numLogDirs, enableAHS); |
| this.overrideClientRMService = overrideClientRMService; |
| this.overrideRTS = overrideRTS; |
| this.overrideApplicationMasterService = overrideApplicationMasterService; |
| } |
| |
| public boolean getStartFailoverFlag() { |
| return startFailover.get(); |
| } |
| |
| public void resetStartFailoverFlag(boolean flag) { |
| startFailover.set(flag); |
| } |
| |
| public void resetFailoverTriggeredFlag(boolean flag) { |
| failoverTriggered.set(flag); |
| } |
| |
| private boolean waittingForFailOver() { |
| int maximumWaittingTime = 50; |
| int count = 0; |
| while (!failoverTriggered.get() && count <= maximumWaittingTime) { |
| try { |
| Thread.sleep(100); |
| } catch (InterruptedException e) { |
| // DO NOTHING |
| } |
| count++; |
| } |
| if (count >= maximumWaittingTime && failoverThread != null) { |
| return false; |
| } |
| try { |
| Thread.sleep(100); |
| } catch (InterruptedException e) { |
| // DO NOTHING |
| } |
| return true; |
| } |
| |
| @Override |
| protected ResourceManager createResourceManager() { |
| return new ResourceManager() { |
| @Override |
| protected void doSecureLogin() throws IOException { |
| // Don't try to login using keytab in the testcases. |
| } |
| @Override |
| protected ClientRMService createClientRMService() { |
| if (overrideClientRMService) { |
| return new CustomedClientRMService(this.rmContext, this.scheduler, |
| this.rmAppManager, this.applicationACLsManager, |
| this.queueACLsManager, |
| this.rmContext.getRMDelegationTokenSecretManager()); |
| } |
| return super.createClientRMService(); |
| } |
| @Override |
| protected ResourceTrackerService createResourceTrackerService() { |
| if (overrideRTS) { |
| return new CustomedResourceTrackerService(this.rmContext, |
| this.nodesListManager, this.nmLivelinessMonitor, |
| this.rmContext.getContainerTokenSecretManager(), |
| this.rmContext.getNMTokenSecretManager()); |
| } |
| return super.createResourceTrackerService(); |
| } |
| @Override |
| protected ApplicationMasterService createApplicationMasterService() { |
| if (overrideApplicationMasterService) { |
| return new CustomedApplicationMasterService(this.rmContext, |
| this.scheduler); |
| } |
| return super.createApplicationMasterService(); |
| } |
| }; |
| } |
| |
| private class CustomedClientRMService extends ClientRMService { |
| public CustomedClientRMService(RMContext rmContext, |
| YarnScheduler scheduler, RMAppManager rmAppManager, |
| ApplicationACLsManager applicationACLsManager, |
| QueueACLsManager queueACLsManager, |
| RMDelegationTokenSecretManager rmDTSecretManager) { |
| super(rmContext, scheduler, rmAppManager, applicationACLsManager, |
| queueACLsManager, rmDTSecretManager); |
| } |
| |
| @Override |
| public GetNewApplicationResponse getNewApplication( |
| GetNewApplicationRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // create the GetNewApplicationResponse with fake applicationId |
| GetNewApplicationResponse response = |
| GetNewApplicationResponse.newInstance( |
| createFakeAppId(), null, null); |
| return response; |
| } |
| |
| @Override |
| public GetApplicationReportResponse getApplicationReport( |
| GetApplicationReportRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // create a fake application report |
| ApplicationReport report = createFakeAppReport(); |
| GetApplicationReportResponse response = |
| GetApplicationReportResponse.newInstance(report); |
| return response; |
| } |
| |
| @Override |
| public GetClusterMetricsResponse getClusterMetrics( |
| GetClusterMetricsRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // create GetClusterMetricsResponse with fake YarnClusterMetrics |
| GetClusterMetricsResponse response = |
| GetClusterMetricsResponse.newInstance( |
| createFakeYarnClusterMetrics()); |
| return response; |
| } |
| |
| @Override |
| public GetApplicationsResponse getApplications( |
| GetApplicationsRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // create GetApplicationsResponse with fake applicationList |
| GetApplicationsResponse response = |
| GetApplicationsResponse.newInstance(createFakeAppReports()); |
| return response; |
| } |
| |
| @Override |
| public GetClusterNodesResponse getClusterNodes( |
| GetClusterNodesRequest request) |
| throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // create GetClusterNodesResponse with fake ClusterNodeLists |
| GetClusterNodesResponse response = |
| GetClusterNodesResponse.newInstance(createFakeNodeReports()); |
| return response; |
| } |
| |
| @Override |
| public GetQueueInfoResponse getQueueInfo(GetQueueInfoRequest request) |
| throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake QueueInfo |
| return GetQueueInfoResponse.newInstance(createFakeQueueInfo()); |
| } |
| |
| @Override |
| public GetQueueUserAclsInfoResponse getQueueUserAcls( |
| GetQueueUserAclsInfoRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake queueUserAcls |
| return GetQueueUserAclsInfoResponse |
| .newInstance(createFakeQueueUserACLInfoList()); |
| } |
| |
| @Override |
| public GetApplicationAttemptReportResponse getApplicationAttemptReport( |
| GetApplicationAttemptReportRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake ApplicationAttemptReport |
| return GetApplicationAttemptReportResponse |
| .newInstance(createFakeApplicationAttemptReport()); |
| } |
| |
| @Override |
| public GetApplicationAttemptsResponse getApplicationAttempts( |
| GetApplicationAttemptsRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake ApplicationAttemptReports |
| return GetApplicationAttemptsResponse |
| .newInstance(createFakeApplicationAttemptReports()); |
| } |
| |
| @Override |
| public GetContainerReportResponse getContainerReport( |
| GetContainerReportRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake containerReport |
| return GetContainerReportResponse |
| .newInstance(createFakeContainerReport()); |
| } |
| |
| @Override |
| public GetContainersResponse getContainers(GetContainersRequest request) |
| throws YarnException, IOException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| // return fake ContainerReports |
| return GetContainersResponse.newInstance(createFakeContainerReports()); |
| } |
| |
| @Override |
| public SubmitApplicationResponse submitApplication( |
| SubmitApplicationRequest request) throws YarnException, IOException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return super.submitApplication(request); |
| } |
| |
| @Override |
| public KillApplicationResponse forceKillApplication( |
| KillApplicationRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return KillApplicationResponse.newInstance(true); |
| } |
| |
| @Override |
| public MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues( |
| MoveApplicationAcrossQueuesRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return Records.newRecord(MoveApplicationAcrossQueuesResponse.class); |
| } |
| |
| @Override |
| public GetDelegationTokenResponse getDelegationToken( |
| GetDelegationTokenRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return GetDelegationTokenResponse.newInstance(createFakeToken()); |
| } |
| |
| @Override |
| public RenewDelegationTokenResponse renewDelegationToken( |
| RenewDelegationTokenRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return RenewDelegationTokenResponse |
| .newInstance(createNextExpirationTime()); |
| } |
| |
| @Override |
| public CancelDelegationTokenResponse cancelDelegationToken( |
| CancelDelegationTokenRequest request) throws YarnException { |
| resetStartFailoverFlag(true); |
| |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| |
| return CancelDelegationTokenResponse.newInstance(); |
| } |
| } |
| |
| public ApplicationReport createFakeAppReport() { |
| ApplicationId appId = ApplicationId.newInstance(1000L, 1); |
| ApplicationAttemptId attemptId = |
| ApplicationAttemptId.newInstance(appId, 1); |
| // create a fake application report |
| ApplicationReport report = |
| ApplicationReport.newInstance(appId, attemptId, "fakeUser", |
| "fakeQueue", "fakeApplicationName", "localhost", 0, null, |
| YarnApplicationState.FINISHED, "fake an application report", "", |
| 1000L, 1000L, 1200L, FinalApplicationStatus.FAILED, null, "", 50f, |
| "fakeApplicationType", null); |
| return report; |
| } |
| |
| public List<ApplicationReport> createFakeAppReports() { |
| List<ApplicationReport> reports = new ArrayList<ApplicationReport>(); |
| reports.add(createFakeAppReport()); |
| return reports; |
| } |
| |
| public ApplicationId createFakeAppId() { |
| return ApplicationId.newInstance(1000L, 1); |
| } |
| |
| public ApplicationAttemptId createFakeApplicationAttemptId() { |
| return ApplicationAttemptId.newInstance(createFakeAppId(), 0); |
| } |
| |
| public ContainerId createFakeContainerId() { |
| return ContainerId.newContainerId(createFakeApplicationAttemptId(), 0); |
| } |
| |
| public YarnClusterMetrics createFakeYarnClusterMetrics() { |
| return YarnClusterMetrics.newInstance(1); |
| } |
| |
| public List<NodeReport> createFakeNodeReports() { |
| NodeId nodeId = NodeId.newInstance("localhost", 0); |
| NodeReport report = |
| NodeReport.newInstance(nodeId, NodeState.RUNNING, "localhost", |
| "rack1", null, null, 4, null, 1000L); |
| List<NodeReport> reports = new ArrayList<NodeReport>(); |
| reports.add(report); |
| return reports; |
| } |
| |
| public QueueInfo createFakeQueueInfo() { |
| return QueueInfo.newInstance("root", 100f, 100f, 50f, null, |
| createFakeAppReports(), QueueState.RUNNING, null, null, null, false, |
| null, false); |
| } |
| |
| public List<QueueUserACLInfo> createFakeQueueUserACLInfoList() { |
| List<QueueACL> queueACL = new ArrayList<QueueACL>(); |
| queueACL.add(QueueACL.SUBMIT_APPLICATIONS); |
| QueueUserACLInfo info = QueueUserACLInfo.newInstance("root", queueACL); |
| List<QueueUserACLInfo> infos = new ArrayList<QueueUserACLInfo>(); |
| infos.add(info); |
| return infos; |
| } |
| |
| public ApplicationAttemptReport createFakeApplicationAttemptReport() { |
| return ApplicationAttemptReport.newInstance( |
| createFakeApplicationAttemptId(), "localhost", 0, "", "", "", |
| YarnApplicationAttemptState.RUNNING, createFakeContainerId(), 1000L, |
| 1200L); |
| } |
| |
| public List<ApplicationAttemptReport> |
| createFakeApplicationAttemptReports() { |
| List<ApplicationAttemptReport> reports = |
| new ArrayList<ApplicationAttemptReport>(); |
| reports.add(createFakeApplicationAttemptReport()); |
| return reports; |
| } |
| |
| public ContainerReport createFakeContainerReport() { |
| return ContainerReport.newInstance(createFakeContainerId(), null, |
| NodeId.newInstance("localhost", 0), null, 1000L, 1200L, "", "", 0, |
| ContainerState.COMPLETE, |
| "http://" + NodeId.newInstance("localhost", 0).toString()); |
| } |
| |
| public List<ContainerReport> createFakeContainerReports() { |
| List<ContainerReport> reports = |
| new ArrayList<ContainerReport>(); |
| reports.add(createFakeContainerReport()); |
| return reports; |
| } |
| |
| public Token createFakeToken() { |
| String identifier = "fake Token"; |
| String password = "fake token passwd"; |
| Token token = Token.newInstance( |
| identifier.getBytes(), " ", password.getBytes(), " "); |
| return token; |
| } |
| |
| public long createNextExpirationTime() { |
| return "fake Token".getBytes().length; |
| } |
| |
| private class CustomedResourceTrackerService extends |
| ResourceTrackerService { |
| public CustomedResourceTrackerService(RMContext rmContext, |
| NodesListManager nodesListManager, |
| NMLivelinessMonitor nmLivelinessMonitor, |
| RMContainerTokenSecretManager containerTokenSecretManager, |
| NMTokenSecretManagerInRM nmTokenSecretManager) { |
| super(rmContext, nodesListManager, nmLivelinessMonitor, |
| containerTokenSecretManager, nmTokenSecretManager); |
| } |
| |
| @Override |
| public RegisterNodeManagerResponse registerNodeManager( |
| RegisterNodeManagerRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| return super.registerNodeManager(request); |
| } |
| |
| @Override |
| public NodeHeartbeatResponse nodeHeartbeat(NodeHeartbeatRequest request) |
| throws YarnException, IOException { |
| resetStartFailoverFlag(true); |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| return super.nodeHeartbeat(request); |
| } |
| } |
| |
| private class CustomedApplicationMasterService extends |
| ApplicationMasterService { |
| public CustomedApplicationMasterService(RMContext rmContext, |
| YarnScheduler scheduler) { |
| super(rmContext, scheduler); |
| } |
| |
| @Override |
| public AllocateResponse allocate(AllocateRequest request) |
| throws YarnException, IOException { |
| resetStartFailoverFlag(true); |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| return createFakeAllocateResponse(); |
| } |
| |
| @Override |
| public RegisterApplicationMasterResponse registerApplicationMaster( |
| RegisterApplicationMasterRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| return createFakeRegisterApplicationMasterResponse(); |
| } |
| |
| @Override |
| public FinishApplicationMasterResponse finishApplicationMaster( |
| FinishApplicationMasterRequest request) throws YarnException, |
| IOException { |
| resetStartFailoverFlag(true); |
| // make sure failover has been triggered |
| Assert.assertTrue(waittingForFailOver()); |
| return createFakeFinishApplicationMasterResponse(); |
| } |
| } |
| |
| public RegisterApplicationMasterResponse |
| createFakeRegisterApplicationMasterResponse() { |
| Resource minCapability = Resource.newInstance(2048, 2); |
| Resource maxCapability = Resource.newInstance(4096, 4); |
| Map<ApplicationAccessType, String> acls = |
| new HashMap<ApplicationAccessType, String>(); |
| acls.put(ApplicationAccessType.MODIFY_APP, "*"); |
| ByteBuffer key = ByteBuffer.wrap("fake_key".getBytes()); |
| return RegisterApplicationMasterResponse.newInstance(minCapability, |
| maxCapability, acls, key, new ArrayList<Container>(), "root", |
| new ArrayList<NMToken>()); |
| } |
| |
| public FinishApplicationMasterResponse |
| createFakeFinishApplicationMasterResponse() { |
| return FinishApplicationMasterResponse.newInstance(true); |
| } |
| |
| public AllocateResponse createFakeAllocateResponse() { |
| if (YarnConfiguration.timelineServiceV2Enabled(getConfig())) { |
| return AllocateResponse.newInstance(-1, |
| new ArrayList<ContainerStatus>(), new ArrayList<Container>(), |
| new ArrayList<NodeReport>(), Resource.newInstance(1024, 2), null, 1, |
| null, new ArrayList<NMToken>(), CollectorInfo.newInstance( |
| "host:port", Token.newInstance(new byte[] {0}, "TIMELINE", |
| new byte[] {0}, "rm"))); |
| } else { |
| return AllocateResponse.newInstance(-1, |
| new ArrayList<ContainerStatus>(), |
| new ArrayList<Container>(), new ArrayList<NodeReport>(), |
| Resource.newInstance(1024, 2), null, 1, |
| null, new ArrayList<NMToken>()); |
| } |
| } |
| } |
| |
| } |
