Skip to content

MAPREDUCE-7407. Avoid stopContainer() on dead node #4779

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
goiri merged 5 commits into from
Sep 15, 2022
Merged

MAPREDUCE-7407. Avoid stopContainer() on dead node #4779

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3abd3f6
MAPREDUCE-7407. Avoid stopContainer() on dead node
Aug 21, 2022
e3baf8b
Moving container check inside switch case and updating Logger
Sep 14, 2022
81bd5b5
removing .toString() from LOG
Sep 15, 2022
1cbbcec
formatting lines
Sep 15, 2022
37be43c
formatting lines
Sep 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 16 additions & 5 deletions ...-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/launcher/ContainerLauncherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,27 +379,38 @@ class EventProcessor implements Runnable {

@Override
public void run() {
LOG.info("Processing the event " + event.toString());
LOG.info("Processing the event {}", event);

// Load ContainerManager tokens before creating a connection.
// TODO: Do it only once per NodeManager.
ContainerId containerID = event.getContainerID();

Container c = getContainer(event);
switch(event.getType()) {

case CONTAINER_REMOTE_LAUNCH:
ContainerRemoteLaunchEvent launchEvent
= (ContainerRemoteLaunchEvent) event;
c.launch(launchEvent);
getContainer(event).launch(launchEvent);
break;

case CONTAINER_REMOTE_CLEANUP:
c.kill(event.getDumpContainerThreads());
// If the container failed to launch earlier (due to dead node for example),
// it has been marked as FAILED and removed from containers during
// CONTAINER_REMOTE_LAUNCH event handling.
// Skip kill() such container during CONTAINER_REMOTE_CLEANUP as
// it is not necessary and could cost 15 minutes delay if the node is dead.
if (!containers.containsKey(containerID)) {
LOG.info("Skip cleanup of already-removed container {}", containerID);
// send killed event to task attempt regardless like in kill().
context.getEventHandler().handle(new TaskAttemptEvent(event.getTaskAttemptID(),
TaskAttemptEventType.TA_CONTAINER_CLEANED));
return;
}
getContainer(event).kill(event.getDumpContainerThreads());
break;

case CONTAINER_COMPLETED:
c.done();
getContainer(event).done();
break;

}
Expand Down
28 changes: 19 additions & 9 deletions .../src/test/java/org/apache/hadoop/mapreduce/v2/app/launcher/TestContainerLauncherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,14 +209,11 @@ public void testHandle() throws Exception {
ut.waitForPoolToIdle();

verify(mockCM).startContainers(any(StartContainersRequest.class));

LOG.info("inserting cleanup event");
ContainerLauncherEvent mockCleanupEvent =
mock(ContainerLauncherEvent.class);
when(mockCleanupEvent.getType())
.thenReturn(EventType.CONTAINER_REMOTE_CLEANUP);
when(mockCleanupEvent.getContainerID())
.thenReturn(contId);
ContainerLauncherEvent mockCleanupEvent = mock(ContainerLauncherEvent.class);
when(mockCleanupEvent.getType()).thenReturn(EventType.CONTAINER_REMOTE_CLEANUP);
when(mockCleanupEvent.getContainerID()).thenReturn(contId);
when(mockCleanupEvent.getTaskAttemptID()).thenReturn(taskAttemptId);
when(mockCleanupEvent.getContainerMgrAddress()).thenReturn(cmAddress);
ut.handle(mockCleanupEvent);
Expand Down Expand Up @@ -283,8 +280,21 @@ public void testOutOfOrder() throws Exception {
ut.handle(mockLaunchEvent);

ut.waitForPoolToIdle();

verify(mockCM, never()).startContainers(any(StartContainersRequest.class));

verify(mockCM).startContainers(any(StartContainersRequest.class));

LOG.info("inserting cleanup event");
ContainerLauncherEvent mockCleanupEvent2 = mock(ContainerLauncherEvent.class);
when(mockCleanupEvent2.getType()).thenReturn(EventType.CONTAINER_REMOTE_CLEANUP);
when(mockCleanupEvent2.getContainerID()).thenReturn(contId);
when(mockCleanupEvent2.getTaskAttemptID()).thenReturn(taskAttemptId);
when(mockCleanupEvent2.getContainerMgrAddress()).thenReturn(cmAddress);
ut.handle(mockCleanupEvent2);

ut.waitForPoolToIdle();

// Verifies stopContainers is called on existing container
verify(mockCM).stopContainers(any(StopContainersRequest.class));
} finally {
ut.stop();
}
Expand Down