You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by jh...@apache.org on 2020/10/28 17:46:04 UTC
[hadoop] branch branch-2.10 updated: YARN-10467. ContainerIdPBImpl
objects can be leaked in RMNodeImpl.completedContainers. Contributed by
Haibo Chen
This is an automated email from the ASF dual-hosted git repository.
jhung pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 6f436a6 YARN-10467. ContainerIdPBImpl objects can be leaked in RMNodeImpl.completedContainers. Contributed by Haibo Chen
6f436a6 is described below
commit 6f436a67766f8e25022b76b228941a69e94c1cc3
Author: Jonathan Hung <jh...@linkedin.com>
AuthorDate: Wed Oct 28 10:45:34 2020 -0700
YARN-10467. ContainerIdPBImpl objects can be leaked in RMNodeImpl.completedContainers. Contributed by Haibo Chen
---
.../rmapp/attempt/RMAppAttemptImpl.java | 30 ++++--
.../rmapp/attempt/TestRMAppAttemptTransitions.java | 115 +++++++++++++++++++++
2 files changed, 138 insertions(+), 7 deletions(-)
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
index 9c6ff33..e565565 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
@@ -170,6 +170,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
private long finishTime = 0;
private long launchAMStartTime = 0;
private long launchAMEndTime = 0;
+ private boolean nonWorkPreservingAMContainerFinished = false;
// Set to null initially. Will eventually get set
// if an RMAppAttemptUnregistrationEvent occurs
@@ -844,7 +845,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
// A new allocate means the AM received the previously sent
// finishedContainers. We can ack this to NM now
- sendFinishedContainersToNM();
+ sendFinishedContainersToNM(finishedContainersSentToAM);
// Mark every containerStatus as being sent to AM though we may return
// only the ones that belong to the current attempt
@@ -1955,13 +1956,14 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
}
// Ack NM to remove finished containers from context.
- private void sendFinishedContainersToNM() {
- for (NodeId nodeId : finishedContainersSentToAM.keySet()) {
+ private void sendFinishedContainersToNM(
+ Map<NodeId, List<ContainerStatus>> finishedContainers) {
+ for (NodeId nodeId : finishedContainers.keySet()) {
// Clear and get current values
List<ContainerStatus> currentSentContainers =
- finishedContainersSentToAM.put(nodeId,
- new ArrayList<ContainerStatus>());
+ finishedContainers.put(nodeId,
+ new ArrayList<ContainerStatus>());
List<ContainerId> containerIdList =
new ArrayList<>(currentSentContainers.size());
for (ContainerStatus containerStatus : currentSentContainers) {
@@ -1970,7 +1972,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
eventHandler.handle(new RMNodeFinishedContainersPulledByAMEvent(nodeId,
containerIdList));
}
- this.finishedContainersSentToAM.clear();
+ finishedContainers.clear();
}
// Add am container to the list so that am container instance will be
@@ -1996,7 +1998,16 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
appAttempt.finishedContainersSentToAM.putIfAbsent(nodeId,
new ArrayList<ContainerStatus>());
appAttempt.finishedContainersSentToAM.get(nodeId).add(containerStatus);
- appAttempt.sendFinishedContainersToNM();
+ appAttempt.sendFinishedContainersToNM(
+ appAttempt.finishedContainersSentToAM);
+ // there might be some completed containers that have not been pulled
+ // by the AM heartbeat, explicitly add them for cleanup.
+ appAttempt.sendFinishedContainersToNM(appAttempt.justFinishedContainers);
+
+ // mark the fact that AM container has finished so that future finished
+ // containers will be cleaned up without the engagement of AM containers
+ // (through heartbeat)
+ appAttempt.nonWorkPreservingAMContainerFinished = true;
} else {
appAttempt.sendFinishedAMContainerToNM(nodeId,
containerStatus.getContainerId());
@@ -2024,6 +2035,11 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
.getNodeId(), new ArrayList<ContainerStatus>());
appAttempt.justFinishedContainers.get(containerFinishedEvent
.getNodeId()).add(containerFinishedEvent.getContainerStatus());
+
+ if (appAttempt.nonWorkPreservingAMContainerFinished) {
+ // AM container has finished, so no more AM heartbeats to do the cleanup.
+ appAttempt.sendFinishedContainersToNM(appAttempt.justFinishedContainers);
+ }
}
private static final class ContainerFinishedAtFinalStateTransition
diff --git a/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 b/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
index 0e4f308..cdbdbe4 100644
--- a/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
+++ b/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
@@ -649,6 +649,8 @@ public class TestRMAppAttemptTransitions {
RMContainer rmContainer = mock(RMContainerImpl.class);
when(scheduler.getRMContainer(container.getId())).
thenReturn(rmContainer);
+ when(container.getNodeId()).thenReturn(
+ BuilderUtils.newNodeId("localhost", 0));
applicationAttempt.handle(
new RMAppAttemptEvent(applicationAttempt.getAppAttemptId(),
@@ -1486,6 +1488,119 @@ public class TestRMAppAttemptTransitions {
.handle(Mockito.any(RMNodeEvent.class));
}
+ /**
+ * Check a completed container that is not yet pulled by AM heartbeat,
+ * is ACKed to NM for cleanup when the AM container exits.
+ */
+ @Test
+ public void testFinishedContainerNotBeingPulledByAMHeartbeat() {
+ Container amContainer = allocateApplicationAttempt();
+ launchApplicationAttempt(amContainer);
+ runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl", false);
+
+ application.handle(new RMAppRunningOnNodeEvent(application
+ .getApplicationId(), amContainer.getNodeId()));
+
+ // Complete a non-AM container
+ ContainerId containerId1 = BuilderUtils.newContainerId(applicationAttempt
+ .getAppAttemptId(), 2);
+ Container container1 = mock(Container.class);
+ ContainerStatus containerStatus1 = mock(ContainerStatus.class);
+ when(container1.getId()).thenReturn(
+ containerId1);
+ when(containerStatus1.getContainerId()).thenReturn(containerId1);
+ when(container1.getNodeId()).thenReturn(NodeId.newInstance("host", 1234));
+ applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
+ applicationAttempt.getAppAttemptId(), containerStatus1,
+ container1.getNodeId()));
+
+ // Verify justFinishedContainers
+ ArgumentCaptor<RMNodeFinishedContainersPulledByAMEvent> captor =
+ ArgumentCaptor.forClass(RMNodeFinishedContainersPulledByAMEvent.class);
+ Assert.assertEquals(1, applicationAttempt.getJustFinishedContainers()
+ .size());
+ Assert.assertEquals(container1.getId(), applicationAttempt
+ .getJustFinishedContainers().get(0).getContainerId());
+ Assert.assertTrue(
+ getFinishedContainersSentToAM(applicationAttempt).isEmpty());
+
+ // finish AM container to emulate AM exit event
+ containerStatus1 = mock(ContainerStatus.class);
+ ContainerId amContainerId = amContainer.getId();
+ when(containerStatus1.getContainerId()).thenReturn(amContainerId);
+ applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
+ applicationAttempt.getAppAttemptId(), containerStatus1,
+ amContainer.getNodeId()));
+
+ Mockito.verify(rmnodeEventHandler, times(2)).handle(captor.capture());
+ List<RMNodeFinishedContainersPulledByAMEvent> containerPulledEvents =
+ captor.getAllValues();
+ // Verify AM container is acked to NM via the RMNodeEvent immediately
+ Assert.assertEquals(amContainer.getId(),
+ containerPulledEvents.get(0).getContainers().get(0));
+ // Verify the non-AM container is acked to NM via the RMNodeEvent
+ Assert.assertEquals(container1.getId(),
+ containerPulledEvents.get(1).getContainers().get(0));
+ Assert.assertTrue("No container shall be added to justFinishedContainers" +
+ " as soon as AM container exits",
+ applicationAttempt.getJustFinishedContainers().isEmpty());
+ Assert.assertTrue(
+ getFinishedContainersSentToAM(applicationAttempt).isEmpty());
+ }
+
+ /**
+ * Check a completed container is ACKed to NM for cleanup after the AM
+ * container has exited.
+ */
+ @Test
+ public void testFinishedContainerAfterAMExit() {
+ Container amContainer = allocateApplicationAttempt();
+ launchApplicationAttempt(amContainer);
+ runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl", false);
+
+ // finish AM container to emulate AM exit event
+ ContainerStatus containerStatus1 = mock(ContainerStatus.class);
+ ContainerId amContainerId = amContainer.getId();
+ when(containerStatus1.getContainerId()).thenReturn(amContainerId);
+ application.handle(new RMAppRunningOnNodeEvent(application
+ .getApplicationId(),
+ amContainer.getNodeId()));
+ applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
+ applicationAttempt.getAppAttemptId(), containerStatus1,
+ amContainer.getNodeId()));
+
+ // Verify AM container is acked to NM via the RMNodeEvent immediately
+ ArgumentCaptor<RMNodeFinishedContainersPulledByAMEvent> captor =
+ ArgumentCaptor.forClass(RMNodeFinishedContainersPulledByAMEvent.class);
+ Mockito.verify(rmnodeEventHandler).handle(captor.capture());
+ Assert.assertEquals(amContainer.getId(),
+ captor.getValue().getContainers().get(0));
+
+ // Complete a non-AM container
+ ContainerId containerId1 = BuilderUtils.newContainerId(applicationAttempt
+ .getAppAttemptId(), 2);
+ Container container1 = mock(Container.class);
+ containerStatus1 = mock(ContainerStatus.class);
+ when(container1.getId()).thenReturn(containerId1);
+ when(containerStatus1.getContainerId()).thenReturn(containerId1);
+ when(container1.getNodeId()).thenReturn(NodeId.newInstance("host", 1234));
+ applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
+ applicationAttempt.getAppAttemptId(), containerStatus1,
+ container1.getNodeId()));
+
+ // Verify container is acked to NM via the RMNodeEvent immediately
+ captor = ArgumentCaptor.forClass(
+ RMNodeFinishedContainersPulledByAMEvent.class);
+ Mockito.verify(rmnodeEventHandler, times(2)).handle(captor.capture());
+ Assert.assertEquals(container1.getId(),
+ captor.getAllValues().get(1).getContainers().get(0));
+ Assert.assertTrue("No container shall be added to justFinishedContainers" +
+ " after AM container exited",
+ applicationAttempt.getJustFinishedContainers().isEmpty());
+ Assert.assertTrue(
+ getFinishedContainersSentToAM(applicationAttempt).isEmpty());
+ }
+
private static List<ContainerStatus> getFinishedContainersSentToAM(
RMAppAttempt applicationAttempt) {
List<ContainerStatus> containers = new ArrayList<ContainerStatus>();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org