You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by vi...@apache.org on 2013/11/18 20:18:12 UTC
svn commit: r1543113 [2/2] - in
/hadoop/common/branches/YARN-321/hadoop-yarn-project: ./
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/
hadoop-yarn/hadoop-yarn-common/src/main/resources/
hadoop-yarn/hadoop-yarn-server/hadoop-yar...
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java Mon Nov 18 19:18:10 2013
@@ -43,6 +43,7 @@ import org.apache.hadoop.yarn.event.Even
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
@@ -99,9 +100,10 @@ public class TestAppManager{
rmDispatcher);
AMLivelinessMonitor amFinishingMonitor = new AMLivelinessMonitor(
rmDispatcher);
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
return new RMContextImpl(rmDispatcher,
containerAllocationExpirer, amLivelinessMonitor, amFinishingMonitor,
- null, null, null, null, null) {
+ null, null, null, null, null, writer) {
@Override
public ConcurrentMap<ApplicationId, RMApp> getRMApps() {
return map;
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java Mon Nov 18 19:18:10 2013
@@ -18,19 +18,19 @@
package org.apache.hadoop.yarn.server.resourcemanager;
+import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyString;
-import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.PrivilegedExceptionAction;
-import java.util.HashSet;
import java.util.EnumSet;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BrokenBarrierException;
@@ -47,10 +47,9 @@ import org.apache.hadoop.security.UserGr
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.MockApps;
import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetQueueInfoResponse;
@@ -78,6 +77,7 @@ import org.apache.hadoop.yarn.factories.
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
@@ -510,6 +510,8 @@ public class TestClientRMService {
.thenReturn(queInfo);
when(yarnScheduler.getQueueInfo(eq("nonexistentqueue"), anyBoolean(), anyBoolean()))
.thenThrow(new IOException("queue does not exist"));
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
+ when(rmContext.getRMApplicationHistoryWriter()).thenReturn(writer);
ConcurrentHashMap<ApplicationId, RMApp> apps = getRMApps(rmContext,
yarnScheduler);
when(rmContext.getRMApps()).thenReturn(apps);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java Mon Nov 18 19:18:10 2013
@@ -101,7 +101,7 @@ public class TestRMNodeTransitions {
rmContext =
new RMContextImpl(rmDispatcher, null, null, null,
- mock(DelegationTokenRenewer.class), null, null, null, null);
+ mock(DelegationTokenRenewer.class), null, null, null, null, null);
scheduler = mock(YarnScheduler.class);
doAnswer(
new Answer<Void>() {
Added: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ahs/TestRMApplicationHistoryWriter.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ahs/TestRMApplicationHistoryWriter.java?rev=1543113&view=auto
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ahs/TestRMApplicationHistoryWriter.java (added)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/ahs/TestRMApplicationHistoryWriter.java Mon Nov 18 19:18:10 2013
@@ -0,0 +1,509 @@
+/**
+ * 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.server.resourcemanager.ahs;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerState;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+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.server.api.protocolrecords.NodeHeartbeatResponse;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.MemoryApplicationHistoryStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.records.ApplicationAttemptHistoryData;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.records.ApplicationHistoryData;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.records.ContainerHistoryData;
+import org.apache.hadoop.yarn.server.resourcemanager.MockAM;
+import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
+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.rmcontainer.RMContainer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestRMApplicationHistoryWriter {
+
+ private static int MAX_RETRIES = 10;
+
+ private RMApplicationHistoryWriter writer;
+ private ApplicationHistoryStore store;
+ private List<CounterDispatcher> dispatchers =
+ new ArrayList<CounterDispatcher>();
+
+ @Before
+ public void setup() {
+ store = new MemoryApplicationHistoryStore();
+ Configuration conf = new Configuration();
+ writer = new RMApplicationHistoryWriter() {
+
+ @Override
+ protected ApplicationHistoryStore createApplicationHistoryStore(
+ Configuration conf) {
+ return store;
+ }
+
+ @Override
+ protected Dispatcher createDispatcher(Configuration conf) {
+ MultiThreadedDispatcher dispatcher = new MultiThreadedDispatcher(conf.getInt(
+ YarnConfiguration.RM_HISTORY_WRITER_MULTI_THREADED_DISPATCHER_POOL_SIZE,
+ YarnConfiguration.DEFAULT_RM_HISTORY_WRITER_MULTI_THREADED_DISPATCHER_POOL_SIZE));
+ dispatcher.setDrainEventsOnStop();
+ return dispatcher;
+ }
+
+ class MultiThreadedDispatcher extends
+ RMApplicationHistoryWriter.MultiThreadedDispatcher {
+
+ public MultiThreadedDispatcher(int num) {
+ super(num);
+ }
+
+ @Override
+ protected AsyncDispatcher createDispatcher() {
+ CounterDispatcher dispatcher = new CounterDispatcher();
+ dispatchers.add(dispatcher);
+ return dispatcher;
+ }
+
+ }
+ };
+ writer.init(conf);
+ writer.start();
+ }
+
+ @After
+ public void tearDown() {
+ writer.stop();
+ }
+
+ private static RMApp createRMApp(ApplicationId appId) {
+ RMApp app = mock(RMApp.class);
+ when(app.getApplicationId()).thenReturn(appId);
+ when(app.getName()).thenReturn("test app");
+ when(app.getApplicationType()).thenReturn("test app type");
+ when(app.getUser()).thenReturn("test user");
+ when(app.getQueue()).thenReturn("test queue");
+ when(app.getSubmitTime()).thenReturn(0L);
+ when(app.getStartTime()).thenReturn(1L);
+ when(app.getFinishTime()).thenReturn(2L);
+ when(app.getDiagnostics()).thenReturn(
+ new StringBuilder("test diagnostics info"));
+ when(app.getFinalApplicationStatus()).thenReturn(
+ FinalApplicationStatus.UNDEFINED);
+ when(app.createApplicationState())
+ .thenReturn(YarnApplicationState.FINISHED);
+ return app;
+ }
+
+ private static RMAppAttempt createRMAppAttempt(
+ ApplicationAttemptId appAttemptId) {
+ RMAppAttempt appAttempt = mock(RMAppAttempt.class);
+ when(appAttempt.getAppAttemptId()).thenReturn(appAttemptId);
+ when(appAttempt.getHost()).thenReturn("test host");
+ when(appAttempt.getRpcPort()).thenReturn(-100);
+ Container container = mock(Container.class);
+ when(container.getId()).thenReturn(
+ ContainerId.newInstance(appAttemptId, 1));
+ when(appAttempt.getMasterContainer()).thenReturn(container);
+ when(appAttempt.getDiagnostics()).thenReturn("test diagnostics info");
+ when(appAttempt.getTrackingUrl()).thenReturn("test url");
+ when(appAttempt.getFinalApplicationStatus()).thenReturn(
+ FinalApplicationStatus.UNDEFINED);
+ when(appAttempt.createApplicationAttemptState()).thenReturn(
+ YarnApplicationAttemptState.FINISHED);
+ return appAttempt;
+ }
+
+ private static RMContainer createRMContainer(
+ ContainerId containerId) {
+ RMContainer container = mock(RMContainer.class);
+ when(container.getContainerId()).thenReturn(containerId);
+ when(container.getAllocatedNode()).thenReturn(
+ NodeId.newInstance("test host", -100));
+ when(container.getAllocatedResource()).thenReturn(
+ Resource.newInstance(-1, -1));
+ when(container.getAllocatedPriority()).thenReturn(Priority.UNDEFINED);
+ when(container.getStartTime()).thenReturn(0L);
+ when(container.getFinishTime()).thenReturn(1L);
+ when(container.getDiagnosticsInfo()).thenReturn("test diagnostics info");
+ when(container.getLogURL()).thenReturn("test log url");
+ when(container.getContainerExitStatus()).thenReturn(-1);
+ when(container.getContainerState()).thenReturn(ContainerState.COMPLETE);
+ return container;
+ }
+
+ @Test
+ public void testWriteApplication() throws Exception {
+ RMApp app = createRMApp(ApplicationId.newInstance(0, 1));
+
+ writer.applicationStarted(app);
+ ApplicationHistoryData appHD = null;
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ appHD = store.getApplication(ApplicationId.newInstance(0, 1));
+ if (appHD != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertNotNull(appHD);
+ Assert.assertEquals("test app", appHD.getApplicationName());
+ Assert.assertEquals("test app type", appHD.getApplicationType());
+ Assert.assertEquals("test user", appHD.getUser());
+ Assert.assertEquals("test queue", appHD.getQueue());
+ Assert.assertEquals(0L, appHD.getSubmitTime());
+ Assert.assertEquals(1L, appHD.getStartTime());
+
+ writer.applicationFinished(app);
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ appHD = store.getApplication(ApplicationId.newInstance(0, 1));
+ if (appHD.getYarnApplicationState() != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertEquals(2L, appHD.getFinishTime());
+ Assert.assertEquals("test diagnostics info", appHD.getDiagnosticsInfo());
+ Assert.assertEquals(FinalApplicationStatus.UNDEFINED,
+ appHD.getFinalApplicationStatus());
+ Assert.assertEquals(YarnApplicationState.FINISHED,
+ appHD.getYarnApplicationState());
+ }
+
+ @Test
+ public void testWriteApplicationAttempt() throws Exception {
+ RMAppAttempt appAttempt = createRMAppAttempt(
+ ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1));
+ writer.applicationAttemptStarted(appAttempt);
+ ApplicationAttemptHistoryData appAttemptHD = null;
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ appAttemptHD =
+ store.getApplicationAttempt(ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1));
+ if (appAttemptHD != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertNotNull(appAttemptHD);
+ Assert.assertEquals("test host", appAttemptHD.getHost());
+ Assert.assertEquals(-100, appAttemptHD.getRPCPort());
+ Assert.assertEquals(
+ ContainerId.newInstance(ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1), 1),
+ appAttemptHD.getMasterContainerId());
+
+ writer.applicationAttemptFinished(appAttempt);
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ appAttemptHD =
+ store.getApplicationAttempt(ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1));
+ if (appAttemptHD.getYarnApplicationAttemptState() != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertEquals("test diagnostics info",
+ appAttemptHD.getDiagnosticsInfo());
+ Assert.assertEquals("test url", appAttemptHD.getTrackingURL());
+ Assert.assertEquals(FinalApplicationStatus.UNDEFINED,
+ appAttemptHD.getFinalApplicationStatus());
+ Assert.assertEquals(YarnApplicationAttemptState.FINISHED,
+ appAttemptHD.getYarnApplicationAttemptState());
+ }
+
+ @Test
+ public void testWriteContainer() throws Exception {
+ RMContainer container = createRMContainer(
+ ContainerId.newInstance(ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1), 1));
+ writer.containerStarted(container);
+ ContainerHistoryData containerHD = null;
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ containerHD =
+ store.getContainer(ContainerId.newInstance(
+ ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1), 1));
+ if (containerHD != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertNotNull(containerHD);
+ Assert.assertEquals(NodeId.newInstance("test host", -100),
+ containerHD.getAssignedNode());
+ Assert.assertEquals(Resource.newInstance(-1, -1),
+ containerHD.getAllocatedResource());
+ Assert.assertEquals(Priority.UNDEFINED, containerHD.getPriority());
+ Assert.assertEquals(0L, container.getStartTime());
+
+ writer.containerFinished(container);
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ containerHD =
+ store.getContainer(ContainerId.newInstance(
+ ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(0, 1), 1), 1));
+ if (containerHD.getContainerState() != null) {
+ break;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ Assert.assertEquals("test diagnostics info",
+ containerHD.getDiagnosticsInfo());
+ Assert.assertEquals("test log url", containerHD.getLogURL());
+ Assert.assertEquals(-1, containerHD.getContainerExitStatus());
+ Assert.assertEquals(ContainerState.COMPLETE,
+ containerHD.getContainerState());
+ }
+
+ @Test
+ public void testParallelWrite() throws Exception {
+ List<ApplicationId> appIds = new ArrayList<ApplicationId>();
+ for (int i = 0; i < 10; ++i) {
+ Random rand = new Random(i);
+ ApplicationId appId = ApplicationId.newInstance(0, rand.nextInt());
+ appIds.add(appId);
+ RMApp app = createRMApp(appId);
+ writer.applicationStarted(app);
+ for (int j = 1; j <= 10; ++j) {
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, j);
+ RMAppAttempt appAttempt = createRMAppAttempt(appAttemptId);
+ writer.applicationAttemptStarted(appAttempt);
+ for (int k = 1; k <= 10; ++k) {
+ ContainerId containerId = ContainerId.newInstance(appAttemptId, k);
+ RMContainer container = createRMContainer(containerId);
+ writer.containerStarted(container);
+ writer.containerFinished(container);
+ }
+ writer.applicationAttemptFinished(appAttempt);
+ }
+ writer.applicationFinished(app);
+ }
+ for (int i = 0; i < MAX_RETRIES; ++i) {
+ if (allEventsHandled(20 * 10 * 10 + 20 * 10 + 20)) {
+ break;
+ } else {
+ Thread.sleep(500);
+ }
+ }
+ Assert.assertTrue(allEventsHandled(20 * 10 * 10 + 20 * 10 + 20));
+ // Validate all events of one application are handled by one dispatcher
+ for (ApplicationId appId : appIds) {
+ Assert.assertTrue(handledByOne(appId));
+ }
+ }
+
+ private boolean allEventsHandled(int expected) {
+ int actual = 0;
+ for (CounterDispatcher dispatcher : dispatchers) {
+ for (Integer count : dispatcher.counts.values()) {
+ actual += count;
+ }
+ }
+ return actual == expected;
+ }
+
+ @Test
+ public void testRMWritingMassiveHistory() throws Exception {
+ // 1. Show RM can run with writing history data
+ // 2. Test additional workload of processing history events
+ YarnConfiguration conf = new YarnConfiguration();
+ // don't process history events
+ MockRM rm = new MockRM(conf) {
+ @Override
+ protected RMApplicationHistoryWriter createRMApplicationHistoryWriter() {
+ return new RMApplicationHistoryWriter() {
+ @Override
+ public void applicationStarted(RMApp app) {
+ }
+ @Override
+ public void applicationFinished(RMApp app) {
+ }
+ @Override
+ public void applicationAttemptStarted(RMAppAttempt appAttempt) {
+ }
+ @Override
+ public void applicationAttemptFinished(RMAppAttempt appAttempt) {
+ }
+ @Override
+ public void containerStarted(RMContainer container) {
+ }
+ @Override
+ public void containerFinished(RMContainer container) {
+ }
+ };
+ }
+ };
+ long startTime1 = System.currentTimeMillis();
+ testRMWritingMassiveHistory(rm);
+ long finishTime1 = System.currentTimeMillis();
+ long elapsedTime1 = finishTime1 - startTime1;
+ rm = new MockRM(conf);
+ long startTime2 = System.currentTimeMillis();
+ testRMWritingMassiveHistory(rm);
+ long finishTime2 = System.currentTimeMillis();
+ long elapsedTime2 = finishTime2 - startTime2;
+ // No more than 10% additional workload
+ // Should be much less, but computation time is fluctuated
+ Assert.assertTrue(elapsedTime2 - elapsedTime1 < elapsedTime1 / 10);
+ }
+
+ private void testRMWritingMassiveHistory(MockRM rm) throws Exception {
+ rm.start();
+ MockNM nm = rm.registerNode("127.0.0.1:1234", 1024 * 10100);
+
+ RMApp app = rm.submitApp(1024);
+ nm.nodeHeartbeat(true);
+ RMAppAttempt attempt = app.getCurrentAppAttempt();
+ MockAM am = rm.sendAMLaunched(attempt.getAppAttemptId());
+ am.registerAppAttempt();
+
+ int request = 10000;
+ am.allocate("127.0.0.1" , 1024, request,
+ new ArrayList<ContainerId>());
+ nm.nodeHeartbeat(true);
+ List<Container> allocated = am.allocate(new ArrayList<ResourceRequest>(),
+ new ArrayList<ContainerId>()).getAllocatedContainers();
+ int waitCount = 0;
+ int allocatedSize = allocated.size();
+ while (allocatedSize < request && waitCount++ < 200) {
+ Thread.sleep(100);
+ allocated = am.allocate(new ArrayList<ResourceRequest>(),
+ new ArrayList<ContainerId>()).getAllocatedContainers();
+ allocatedSize += allocated.size();
+ nm.nodeHeartbeat(true);
+ }
+ Assert.assertEquals(request, allocatedSize);
+
+ am.unregisterAppAttempt();
+ am.waitForState(RMAppAttemptState.FINISHING);
+ nm.nodeHeartbeat(am.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
+ am.waitForState(RMAppAttemptState.FINISHED);
+
+ NodeHeartbeatResponse resp = nm.nodeHeartbeat(true);
+ List<ContainerId> cleaned = resp.getContainersToCleanup();
+ int cleanedSize = cleaned.size();
+ waitCount = 0;
+ while (cleanedSize < allocatedSize && waitCount++ < 200) {
+ Thread.sleep(100);
+ resp = nm.nodeHeartbeat(true);
+ cleaned = resp.getContainersToCleanup();
+ cleanedSize += cleaned.size();
+ }
+ Assert.assertEquals(allocatedSize, cleanedSize);
+ rm.waitForState(app.getApplicationId(), RMAppState.FINISHED);
+
+ rm.stop();
+ }
+
+ private boolean handledByOne(ApplicationId appId) {
+ int count = 0;
+ for (CounterDispatcher dispatcher : dispatchers) {
+ if (dispatcher.counts.containsKey(appId)) {
+ ++count;
+ }
+ }
+ return count == 1;
+ }
+
+ private static class CounterDispatcher extends AsyncDispatcher {
+
+ private Map<ApplicationId, Integer> counts =
+ new HashMap<ApplicationId, Integer>();
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ protected void dispatch(Event event) {
+ if (event instanceof WritingApplicationHistoryEvent) {
+ WritingApplicationHistoryEvent ashEvent =
+ (WritingApplicationHistoryEvent) event;
+ switch (ashEvent.getType()) {
+ case APP_START:
+ incrementCounts(((WritingApplicationStartEvent) event).getApplicationId());
+ break;
+ case APP_FINISH:
+ incrementCounts(((WritingApplicationFinishEvent) event)
+ .getApplicationId());
+ break;
+ case APP_ATTEMPT_START:
+ incrementCounts(((WritingApplicationAttemptStartEvent) event)
+ .getApplicationAttemptId().getApplicationId());
+ break;
+ case APP_ATTEMPT_FINISH:
+ incrementCounts(((WritingApplicationAttemptFinishEvent) event)
+ .getApplicationAttemptId().getApplicationId());
+ break;
+ case CONTAINER_START:
+ incrementCounts(((WritingContainerStartEvent) event).getContainerId()
+ .getApplicationAttemptId().getApplicationId());
+ break;
+ case CONTAINER_FINISH:
+ incrementCounts(((WritingContainerFinishEvent) event).getContainerId()
+ .getApplicationAttemptId().getApplicationId());
+ break;
+ }
+ }
+ super.dispatch(event);
+ }
+
+ private void incrementCounts(ApplicationId appId) {
+ Integer val = counts.get(appId);
+ if (val == null) {
+ counts.put(appId, 1);
+ } else {
+ counts.put(appId, val + 1);
+ }
+ }
+ }
+
+}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMExpiry.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMExpiry.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMExpiry.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMExpiry.java Mon Nov 18 19:18:10 2013
@@ -72,7 +72,7 @@ public class TestNMExpiry {
// Dispatcher that processes events inline
Dispatcher dispatcher = new InlineDispatcher();
RMContext context = new RMContextImpl(dispatcher, null,
- null, null, null, null, null, null, null);
+ null, null, null, null, null, null, null, null);
dispatcher.register(SchedulerEventType.class,
new InlineDispatcher.EmptyEventHandler());
dispatcher.register(RMNodeEventType.class,
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMReconnect.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMReconnect.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMReconnect.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestNMReconnect.java Mon Nov 18 19:18:10 2013
@@ -71,7 +71,7 @@ public class TestNMReconnect {
new TestRMNodeEventDispatcher());
RMContext context = new RMContextImpl(dispatcher, null,
- null, null, null, null, null, null, null);
+ null, null, null, null, null, null, null, null);
dispatcher.register(SchedulerEventType.class,
new InlineDispatcher.EmptyEventHandler());
dispatcher.register(RMNodeEventType.class,
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestRMNMRPCResponseId.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestRMNMRPCResponseId.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestRMNMRPCResponseId.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/resourcetracker/TestRMNMRPCResponseId.java Mon Nov 18 19:18:10 2013
@@ -71,7 +71,7 @@ public class TestRMNMRPCResponseId {
RMContext context =
new RMContextImpl(dispatcher, null, null, null, null,
null, new RMContainerTokenSecretManager(conf),
- new NMTokenSecretManagerInRM(conf), null);
+ new NMTokenSecretManagerInRM(conf), null, null);
dispatcher.register(RMNodeEventType.class,
new ResourceManager.NodeEventDispatcher(context));
NodesListManager nodesListManager = new NodesListManager(context);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java Mon Nov 18 19:18:10 2013
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertEqu
import static org.junit.Assume.assumeTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -57,6 +58,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
@@ -95,6 +97,7 @@ public class TestRMAppTransitions {
private static int appId = 1;
private DrainDispatcher rmDispatcher;
private RMStateStore store;
+ private RMApplicationHistoryWriter writer;
private YarnScheduler scheduler;
// ignore all the RM application attempt events
@@ -190,13 +193,15 @@ public class TestRMAppTransitions {
AMLivelinessMonitor amLivelinessMonitor = mock(AMLivelinessMonitor.class);
AMLivelinessMonitor amFinishingMonitor = mock(AMLivelinessMonitor.class);
store = mock(RMStateStore.class);
+ writer = mock(RMApplicationHistoryWriter.class);
this.rmContext =
new RMContextImpl(rmDispatcher, store,
containerAllocationExpirer, amLivelinessMonitor, amFinishingMonitor,
null, new AMRMTokenSecretManager(conf),
new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM());
+ new ClientToAMTokenSecretManagerInRM(),
+ writer);
rmDispatcher.register(RMAppAttemptEventType.class,
new TestApplicationAttemptEventDispatcher(this.rmContext));
@@ -345,6 +350,7 @@ public class TestRMAppTransitions {
protected RMApp testCreateAppNewSaving(
ApplicationSubmissionContext submissionContext) throws IOException {
RMApp application = createNewTestApp(submissionContext);
+ verify(writer).applicationStarted(any(RMApp.class));
// NEW => NEW_SAVING event RMAppEventType.START
RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.START);
@@ -466,6 +472,9 @@ public class TestRMAppTransitions {
Assert.assertTrue("Finished app missing diagnostics",
application.getDiagnostics().indexOf(diagMsg) != -1);
+ // reset the counter of Mockito.verify
+ reset(writer);
+
// test app fails after 1 app attempt failure
LOG.info("--- START: testUnmanagedAppFailPath ---");
application = testCreateAppRunning(subContext);
@@ -505,6 +514,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertKilled(application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -519,6 +529,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertFailed(application, rejectedText);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test (timeout = 30000)
@@ -532,6 +543,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertKilled(application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test (timeout = 30000)
@@ -546,6 +558,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertFailed(application, rejectedText);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test (timeout = 30000)
@@ -560,6 +573,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertFailed(application, rejectedText);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -572,6 +586,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertAppAndAttemptKilled(application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -604,6 +619,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertFailed(application, ".*" + message + ".*Failing the application.*");
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -616,6 +632,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertAppAndAttemptKilled(application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -637,6 +654,7 @@ public class TestRMAppTransitions {
assertAppState(RMAppState.FINAL_SAVING, application);
assertKilled(application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -688,6 +706,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertFailed(application, ".*Failing the application.*");
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -701,6 +720,7 @@ public class TestRMAppTransitions {
application.handle(event);
rmDispatcher.await();
assertAppState(RMAppState.FINISHED, application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
// While App is at FINAL_SAVING, Attempt_Finished event may come before
@@ -745,6 +765,7 @@ public class TestRMAppTransitions {
StringBuilder diag = application.getDiagnostics();
Assert.assertEquals("application diagnostics is not correct",
"", diag.toString());
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test (timeout = 30000)
@@ -775,6 +796,7 @@ public class TestRMAppTransitions {
rmDispatcher.await();
assertTimesAtFinish(application);
assertAppState(RMAppState.FAILED, application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test (timeout = 30000)
@@ -830,6 +852,7 @@ public class TestRMAppTransitions {
rmDispatcher.await();
assertTimesAtFinish(application);
assertAppState(RMAppState.KILLED, application);
+ verify(writer).applicationFinished(any(RMApp.class));
}
@Test
@@ -853,6 +876,9 @@ public class TestRMAppTransitions {
report = app.createAndGetApplicationReport("clientuser", true);
Assert.assertNull(report.getClientToAMToken());
+ // reset the counter of Mockito.verify
+ reset(writer);
+
app = testCreateAppRunning(null);
rmDispatcher.await();
assertAppState(RMAppState.RUNNING, app);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java Mon Nov 18 19:18:10 2013
@@ -60,6 +60,7 @@ import org.apache.hadoop.yarn.event.Inli
import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEvent;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.ApplicationMasterLauncher;
@@ -116,6 +117,8 @@ public class TestRMAppAttemptTransitions
private ApplicationMasterLauncher applicationMasterLauncher;
private AMLivelinessMonitor amLivelinessMonitor;
private AMLivelinessMonitor amFinishingMonitor;
+ private RMApplicationHistoryWriter writer;
+
private RMStateStore store;
private RMApp application;
@@ -204,13 +207,15 @@ public class TestRMAppAttemptTransitions
mock(ContainerAllocationExpirer.class);
amLivelinessMonitor = mock(AMLivelinessMonitor.class);
amFinishingMonitor = mock(AMLivelinessMonitor.class);
+ writer = mock(RMApplicationHistoryWriter.class);
rmContext =
new RMContextImpl(rmDispatcher,
containerAllocationExpirer, amLivelinessMonitor, amFinishingMonitor,
null, amRMTokenManager,
new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- clientToAMTokenManager);
+ clientToAMTokenManager,
+ writer);
store = mock(RMStateStore.class);
((RMContextImpl) rmContext).setStateStore(store);
@@ -368,6 +373,7 @@ public class TestRMAppAttemptTransitions
assertEquals(0, applicationAttempt.getRanNodes().size());
assertNull(applicationAttempt.getFinalApplicationStatus());
verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
+ verify(writer).applicationAttemptFinished(any(RMAppAttempt.class));
verifyAttemptFinalStateSaved();
}
@@ -445,6 +451,7 @@ public class TestRMAppAttemptTransitions
// Check events
verify(application, times(2)).handle(any(RMAppFailedAttemptEvent.class));
verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
+ verify(writer).applicationAttemptFinished(any(RMAppAttempt.class));
verifyAttemptFinalStateSaved();
}
@@ -480,6 +487,7 @@ public class TestRMAppAttemptTransitions
assertEquals(getProxyUrl(applicationAttempt),
applicationAttempt.getTrackingUrl());
}
+ verify(writer).applicationAttemptStarted(any(RMAppAttempt.class));
// TODO - need to add more checks relevant to this state
}
@@ -526,6 +534,7 @@ public class TestRMAppAttemptTransitions
assertEquals(container, applicationAttempt.getMasterContainer());
assertEquals(finalStatus, applicationAttempt.getFinalApplicationStatus());
verifyTokenCount(applicationAttempt.getAppAttemptId(), 1);
+ verify(writer).applicationAttemptFinished(any(RMAppAttempt.class));
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/TestRMContainerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/TestRMContainerImpl.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/TestRMContainerImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/TestRMContainerImpl.java Mon Nov 18 19:18:10 2013
@@ -19,9 +19,12 @@
package org.apache.hadoop.yarn.server.resourcemanager.rmcontainer;
import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -36,6 +39,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.DrainDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptContainerFinishedEvent;
@@ -48,12 +53,10 @@ import org.mockito.ArgumentCaptor;
@SuppressWarnings({ "unchecked", "rawtypes" })
public class TestRMContainerImpl {
- @SuppressWarnings("resource")
@Test
public void testReleaseWhileRunning() {
DrainDispatcher drainDispatcher = new DrainDispatcher();
- EventHandler eventHandler = drainDispatcher.getEventHandler();
EventHandler<RMAppAttemptEvent> appAttemptEventHandler = mock(EventHandler.class);
EventHandler generic = mock(EventHandler.class);
drainDispatcher.register(RMAppAttemptEventType.class,
@@ -74,19 +77,24 @@ public class TestRMContainerImpl {
Container container = BuilderUtils.newContainer(containerId, nodeId,
"host:3465", resource, priority, null);
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
+ RMContext rmContext = mock(RMContext.class);
+ when(rmContext.getDispatcher()).thenReturn(drainDispatcher);
+ when(rmContext.getContainerAllocationExpirer()).thenReturn(expirer);
+ when(rmContext.getRMApplicationHistoryWriter()).thenReturn(writer);
RMContainer rmContainer = new RMContainerImpl(container, appAttemptId,
- nodeId, eventHandler, expirer, "user");
+ nodeId, "user", rmContext);
assertEquals(RMContainerState.NEW, rmContainer.getState());
assertEquals(resource, rmContainer.getAllocatedResource());
assertEquals(nodeId, rmContainer.getAllocatedNode());
assertEquals(priority, rmContainer.getAllocatedPriority());
+ verify(writer).containerStarted(any(RMContainer.class));
rmContainer.handle(new RMContainerEvent(containerId,
RMContainerEventType.START));
drainDispatcher.await();
assertEquals(RMContainerState.ALLOCATED, rmContainer.getState());
-
rmContainer.handle(new RMContainerEvent(containerId,
RMContainerEventType.ACQUIRED));
drainDispatcher.await();
@@ -114,6 +122,7 @@ public class TestRMContainerImpl {
assertEquals(ContainerExitStatus.ABORTED,
rmContainer.getContainerExitStatus());
assertEquals(ContainerState.COMPLETE, rmContainer.getContainerState());
+ verify(writer).containerFinished(any(RMContainer.class));
ArgumentCaptor<RMAppAttemptContainerFinishedEvent> captor = ArgumentCaptor
.forClass(RMAppAttemptContainerFinishedEvent.class);
@@ -130,12 +139,10 @@ public class TestRMContainerImpl {
assertEquals(RMContainerState.RELEASED, rmContainer.getState());
}
- @SuppressWarnings("resource")
@Test
public void testExpireWhileRunning() {
DrainDispatcher drainDispatcher = new DrainDispatcher();
- EventHandler eventHandler = drainDispatcher.getEventHandler();
EventHandler<RMAppAttemptEvent> appAttemptEventHandler = mock(EventHandler.class);
EventHandler generic = mock(EventHandler.class);
drainDispatcher.register(RMAppAttemptEventType.class,
@@ -156,13 +163,19 @@ public class TestRMContainerImpl {
Container container = BuilderUtils.newContainer(containerId, nodeId,
"host:3465", resource, priority, null);
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
+ RMContext rmContext = mock(RMContext.class);
+ when(rmContext.getDispatcher()).thenReturn(drainDispatcher);
+ when(rmContext.getContainerAllocationExpirer()).thenReturn(expirer);
+ when(rmContext.getRMApplicationHistoryWriter()).thenReturn(writer);
RMContainer rmContainer = new RMContainerImpl(container, appAttemptId,
- nodeId, eventHandler, expirer, "user");
+ nodeId, "user", rmContext);
assertEquals(RMContainerState.NEW, rmContainer.getState());
assertEquals(resource, rmContainer.getAllocatedResource());
assertEquals(nodeId, rmContainer.getAllocatedNode());
assertEquals(priority, rmContainer.getAllocatedPriority());
+ verify(writer).containerStarted(any(RMContainer.class));
rmContainer.handle(new RMContainerEvent(containerId,
RMContainerEventType.START));
@@ -191,5 +204,6 @@ public class TestRMContainerImpl {
containerStatus, RMContainerEventType.EXPIRE));
drainDispatcher.await();
assertEquals(RMContainerState.RUNNING, rmContainer.getState());
+ verify(writer, never()).containerFinished(any(RMContainer.class));
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java Mon Nov 18 19:18:10 2013
@@ -345,7 +345,7 @@ public class TestCapacityScheduler {
cs.reinitialize(conf, new RMContextImpl(null, null, null, null, null,
null, new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
checkQueueCapacities(cs, A_CAPACITY, B_CAPACITY);
conf.setCapacity(A, 80f);
@@ -444,7 +444,7 @@ public class TestCapacityScheduler {
cs.reinitialize(conf, new RMContextImpl(null, null, null, null, null,
null, new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
}
@Test
@@ -457,7 +457,7 @@ public class TestCapacityScheduler {
cs.reinitialize(csConf, new RMContextImpl(null, null, null, null,
null, null, new RMContainerTokenSecretManager(csConf),
new NMTokenSecretManagerInRM(csConf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
RMNode n1 = MockNodes.newNodeInfo(0, MockNodes.newResource(4 * GB), 1);
RMNode n2 = MockNodes.newNodeInfo(0, MockNodes.newResource(2 * GB), 2);
@@ -484,7 +484,7 @@ public class TestCapacityScheduler {
cs.reinitialize(conf, new RMContextImpl(null, null, null, null, null,
null, new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
checkQueueCapacities(cs, A_CAPACITY, B_CAPACITY);
// Add a new queue b4
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestChildQueueOrder.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestChildQueueOrder.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestChildQueueOrder.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestChildQueueOrder.java Mon Nov 18 19:18:10 2013
@@ -41,8 +41,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.DrainDispatcher;
-import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
@@ -248,14 +248,18 @@ public class TestChildQueueOrder {
ContainerAllocationExpirer expirer =
mock(ContainerAllocationExpirer.class);
DrainDispatcher drainDispatcher = new DrainDispatcher();
- EventHandler eventHandler = drainDispatcher.getEventHandler();
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
+ RMContext rmContext = mock(RMContext.class);
+ when(rmContext.getContainerAllocationExpirer()).thenReturn(expirer);
+ when(rmContext.getDispatcher()).thenReturn(drainDispatcher);
+ when(rmContext.getRMApplicationHistoryWriter()).thenReturn(writer);
ApplicationAttemptId appAttemptId = BuilderUtils.newApplicationAttemptId(
app_0.getApplicationId(), 1);
ContainerId containerId = BuilderUtils.newContainerId(appAttemptId, 1);
Container container=TestUtils.getMockContainer(containerId,
node_0.getNodeID(), Resources.createResource(1*GB), priority);
RMContainer rmContainer = new RMContainerImpl(container, appAttemptId,
- node_0.getNodeID(), eventHandler, expirer, "user");
+ node_0.getNodeID(), "user", rmContext);
// Assign {1,2,3,4} 1GB containers respectively to queues
stubQueueAllocation(a, clusterResource, node_0, 1*GB);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestQueueParsing.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestQueueParsing.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestQueueParsing.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestQueueParsing.java Mon Nov 18 19:18:10 2013
@@ -47,7 +47,7 @@ public class TestQueueParsing {
capacityScheduler.reinitialize(conf, new RMContextImpl(null, null,
null, null, null, null, new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
CSQueue a = capacityScheduler.getQueue("a");
Assert.assertEquals(0.10, a.getAbsoluteCapacity(), DELTA);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java Mon Nov 18 19:18:10 2013
@@ -41,6 +41,7 @@ import org.apache.hadoop.yarn.factories.
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
@@ -84,12 +85,13 @@ public class TestUtils {
new ContainerAllocationExpirer(nullDispatcher);
Configuration conf = new Configuration();
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
RMContext rmContext =
new RMContextImpl(nullDispatcher, cae, null, null, null,
new AMRMTokenSecretManager(conf),
new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM());
+ new ClientToAMTokenSecretManagerInRM(), writer);
return rmContext;
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java Mon Nov 18 19:18:10 2013
@@ -20,6 +20,8 @@ package org.apache.hadoop.yarn.server.re
import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
@@ -55,6 +57,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
import org.apache.hadoop.yarn.server.resourcemanager.Task;
+import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
@@ -136,8 +139,9 @@ public class TestFifoScheduler {
@Test(timeout=5000)
public void testAppAttemptMetrics() throws Exception {
AsyncDispatcher dispatcher = new InlineDispatcher();
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
RMContext rmContext = new RMContextImpl(dispatcher, null,
- null, null, null, null, null, null, null);
+ null, null, null, null, null, null, null, writer);
FifoScheduler schedular = new FifoScheduler();
schedular.reinitialize(new Configuration(), rmContext);
@@ -169,8 +173,9 @@ public class TestFifoScheduler {
NMTokenSecretManagerInRM nmTokenSecretManager =
new NMTokenSecretManagerInRM(conf);
nmTokenSecretManager.rollMasterKey();
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
RMContext rmContext = new RMContextImpl(dispatcher, null, null, null, null,
- null, containerTokenSecretManager, nmTokenSecretManager, null);
+ null, containerTokenSecretManager, nmTokenSecretManager, null, writer);
FifoScheduler scheduler = new FifoScheduler();
scheduler.reinitialize(new Configuration(), rmContext);
@@ -229,8 +234,9 @@ public class TestFifoScheduler {
NMTokenSecretManagerInRM nmTokenSecretManager =
new NMTokenSecretManagerInRM(conf);
nmTokenSecretManager.rollMasterKey();
+ RMApplicationHistoryWriter writer = mock(RMApplicationHistoryWriter.class);
RMContext rmContext = new RMContextImpl(dispatcher, null, null, null, null,
- null, containerTokenSecretManager, nmTokenSecretManager, null);
+ null, containerTokenSecretManager, nmTokenSecretManager, null, writer);
FifoScheduler scheduler = new FifoScheduler(){
@SuppressWarnings("unused")
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebApp.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebApp.java?rev=1543113&r1=1543112&r2=1543113&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebApp.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebApp.java Mon Nov 18 19:18:10 2013
@@ -161,7 +161,7 @@ public class TestRMWebApp {
deactivatedNodesMap.put(node.getHostName(), node);
}
return new RMContextImpl(null, null, null, null,
- null, null, null, null, null) {
+ null, null, null, null, null, null) {
@Override
public ConcurrentMap<ApplicationId, RMApp> getRMApps() {
return applicationsMaps;
@@ -204,7 +204,7 @@ public class TestRMWebApp {
cs.reinitialize(conf, new RMContextImpl(null, null, null, null, null,
null, new RMContainerTokenSecretManager(conf),
new NMTokenSecretManagerInRM(conf),
- new ClientToAMTokenSecretManagerInRM()));
+ new ClientToAMTokenSecretManagerInRM(), null));
return cs;
}