pan3793 commented on code in PR #8691:
URL: https://github.com/apache/hadoop/pull/8691#discussion_r3931739121
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java:
##########
@@ -1294,8 +1294,17 @@ public void run() {
// On error, report failure to Container and signal ABORT
// Notify resource of failed localization
ContainerId cId = context.getContainerId();
- dispatcher.getEventHandler().handle(new ContainerResourceFailedEvent(
- cId, null, exception.getMessage()));
+ try {
+ dispatcher.getEventHandler().handle(new
ContainerResourceFailedEvent(
+ cId, null, exception.getMessage()));
+ } catch (Exception e) {
+ LOG.warn("Failed to send container resource failed event for {}",
+ cId, e);
+ if (e instanceof InterruptedException
+ || e.getCause() instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
}
List<Path> paths = new ArrayList<Path>();
for (LocalizerResourceRequestEvent event : scheduled.values()) {
Review Comment:
`scheduled` is a plain HashMap. The IPC handler mutates it and unlocks the
same resources in `processHeartbeat` under `synchronized (privLocalizers)`
(L1125, L1188-L1189, L1202-L1203); this loop runs on the runner thread without
that monitor. `cleanupPrivLocalizers` removes then interrupts under the
monitor, but `LocalizerTracker.serviceStop` interrupts without it, and under
LCE a localizer process that dies mid-heartbeat leaves the handler in
`processHeartbeat` while the runner enters this finally. A CME here skips the
remaining `unlock()` calls and both `delete()` calls, which is the leak this PR
fixes. Since the PR makes this loop reachable on the interrupt path, please
take the monitor here as well:
```java
List<Path> paths = new ArrayList<Path>();
synchronized (localizerTracker.privLocalizers) {
for (LocalizerResourceRequestEvent event : scheduled.values()) {
...
event.getResource().unlock();
}
scheduled.clear();
}
```
`scheduled.clear()` also stops a late heartbeat from releasing a Semaphore
the loop already released; `processHeartbeat` already handles a missing entry
as "Unknown resource reported".
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java:
##########
@@ -1294,8 +1294,17 @@ public void run() {
// On error, report failure to Container and signal ABORT
// Notify resource of failed localization
ContainerId cId = context.getContainerId();
- dispatcher.getEventHandler().handle(new ContainerResourceFailedEvent(
- cId, null, exception.getMessage()));
+ try {
+ dispatcher.getEventHandler().handle(new
ContainerResourceFailedEvent(
+ cId, null, exception.getMessage()));
+ } catch (Exception e) {
+ LOG.warn("Failed to send container resource failed event for {}",
+ cId, e);
+ if (e instanceof InterruptedException
+ || e.getCause() instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
Review Comment:
`handle()` declares no checked exception, so `e instanceof
InterruptedException` is dead; only the `getCause()` branch is live.
`AsyncDispatcher.GenericEventHandler.handle` already logs this throwable with a
stack at WARN before throwing, so this WARN prints the same stack a second time
on every kill. Also worth stating that not delivering the event on interrupt is
intentional: the re-init path in `handleInitContainerResources` interrupts the
old runner while the container is REINITIALIZING, and `REINITIALIZING +
RESOURCE_FAILED -> RUNNING` would abort the re-init.
```suggestion
} catch (Exception e) {
if (e instanceof YarnRuntimeException
&& e.getCause() instanceof InterruptedException) {
// Interrupted by container kill or re-init. The event is
// intentionally not delivered: in REINITIALIZING,
RESOURCE_FAILED
// would abort the re-init. AsyncDispatcher already logged the
// stack trace.
LOG.info("Localizer {} interrupted, skip sending resource
failed"
+ " event for {}", localizerId, cId);
Thread.currentThread().interrupt();
} else {
LOG.error("Failed to send resource failed event for {}", cId,
e);
}
}
```
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java:
##########
@@ -818,6 +822,76 @@ public void testLocalizerRunnerException() throws
Exception {
}
}
+ @Test
+ @Timeout(value = 10)
+ @SuppressWarnings("unchecked") // mocked generics
+ public void testDownloadingResourcesCleanedUpWhenDispatchFails()
+ throws Exception {
+ Dispatcher dispatcher = mock(Dispatcher.class);
Review Comment:
The mocked dispatcher hand-simulates `AsyncDispatcher`'s contract, so the
test keeps passing if the dispatcher stops throwing or starts re-interrupting,
and `assertTrue(Thread.interrupted())` then only proves the catch block ran. A
real `DrainDispatcher` with the flag set before `run()` exercises the
production path deterministically: `LinkedBlockingQueue.put` throws on entry
and clears the flag, `GenericEventHandler.handle` wraps it as
`YarnRuntimeException`, and the assertion proves the flag was restored.
```java
DrainDispatcher dispatcher = new DrainDispatcher();
dispatcher.init(conf);
dispatcher.start();
try {
...
Thread.currentThread().interrupt();
runner.run();
...
} finally {
dispatcher.stop();
}
```
This also drops the `Dispatcher`, `Event` and `YarnRuntimeException` imports
and the method-level `@SuppressWarnings`.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java:
##########
@@ -818,6 +822,76 @@ public void testLocalizerRunnerException() throws
Exception {
}
}
+ @Test
+ @Timeout(value = 10)
+ @SuppressWarnings("unchecked") // mocked generics
+ public void testDownloadingResourcesCleanedUpWhenDispatchFails()
+ throws Exception {
+ Dispatcher dispatcher = mock(Dispatcher.class);
+ EventHandler<Event> eventHandler = mock(EventHandler.class);
+ when(dispatcher.getEventHandler()).thenReturn(eventHandler);
+ // Simulate the localizer thread being interrupted by a container kill:
+ // dispatching the failure event throws instead of completing.
+ Mockito.doThrow(new YarnRuntimeException(new InterruptedException()))
+ .when(eventHandler).handle(isA(ContainerResourceFailedEvent.class));
+
+ ContainerExecutor exec = mock(ContainerExecutor.class);
+ DeletionService delService = mock(DeletionService.class);
+ LocalDirsHandlerService dirsHandlerSpy = spy(new
LocalDirsHandlerService());
+ dirsHandlerSpy.init(conf);
Review Comment:
`spy` + `init(conf)` runs the real `serviceInit` (mkdirs, DiskChecker, disk
usage probes) and creates `${hadoop.tmp.dir}/nm-local-dir` outside `basedir`,
which `cleanup()` does not remove. The only call on the tested path is the
stubbed `getLocalPathForWrite`, so `mock(LocalDirsHandlerService.class)`
without `init` is equivalent, as at L3012 and L3268.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java:
##########
@@ -818,6 +822,76 @@ public void testLocalizerRunnerException() throws
Exception {
}
}
+ @Test
+ @Timeout(value = 10)
+ @SuppressWarnings("unchecked") // mocked generics
+ public void testDownloadingResourcesCleanedUpWhenDispatchFails()
+ throws Exception {
+ Dispatcher dispatcher = mock(Dispatcher.class);
+ EventHandler<Event> eventHandler = mock(EventHandler.class);
+ when(dispatcher.getEventHandler()).thenReturn(eventHandler);
+ // Simulate the localizer thread being interrupted by a container kill:
+ // dispatching the failure event throws instead of completing.
+ Mockito.doThrow(new YarnRuntimeException(new InterruptedException()))
+ .when(eventHandler).handle(isA(ContainerResourceFailedEvent.class));
+
+ ContainerExecutor exec = mock(ContainerExecutor.class);
+ DeletionService delService = mock(DeletionService.class);
+ LocalDirsHandlerService dirsHandlerSpy = spy(new
LocalDirsHandlerService());
+ dirsHandlerSpy.init(conf);
+ // Fail localization so LocalizerRunner.run() takes the error path.
+ Mockito.doThrow(new IOException("Simulated disk failure"))
+ .when(dirsHandlerSpy).getLocalPathForWrite(isA(String.class));
+
+ ResourceLocalizationService rls =
+ new ResourceLocalizationService(dispatcher, exec, delService,
+ dirsHandlerSpy, nmContext, metrics);
+
+ final ApplicationId appId =
+ BuilderUtils.newApplicationId(314159265358979L, 3);
+ final Container c = getMockContainer(appId, 42, "user0");
+ LocalizerRunner runner = rls.new LocalizerRunner(
+ new LocalizerContext("user0", c.getContainerId(), null),
+ c.getContainerId().toString());
+
+ // A resource that was in DOWNLOADING state when the localizer died.
+ LocalizedResource rsrc = mock(LocalizedResource.class);
+ when(rsrc.getLocalPath()).thenReturn(
+ new Path("/local/usercache/user0/filecache/10/foo.jar"));
+ LocalizerResourceRequestEvent scheduledEvent =
+ mock(LocalizerResourceRequestEvent.class);
+ when(scheduledEvent.getResource()).thenReturn(rsrc);
+ runner.scheduled.put(mock(LocalResourceRequest.class), scheduledEvent);
+
+ // Must not propagate the dispatch failure, and must still unlock the
+ // DOWNLOADING resource and schedule the deletion tasks.
+ runner.run();
+
+ // The interrupt status must be restored after the dispatch failure was
+ // swallowed. Thread.interrupted() also clears it for the test thread.
+ assertTrue(Thread.interrupted());
+
+ verify(rsrc).unlock();
+ ArgumentCaptor<FileDeletionTask> captor =
+ ArgumentCaptor.forClass(FileDeletionTask.class);
+ verify(delService, times(2)).delete(captor.capture());
+ List<FileDeletionTask> tasks = captor.getAllValues();
+ // Localization dir and _tmp download dir of the DOWNLOADING resource.
+ FileDeletionTask rsrcTask = tasks.get(0);
+ assertEquals("user0", rsrcTask.getUser());
+ assertNull(rsrcTask.getSubDir());
+ assertEquals(Arrays.asList(
+ new Path("/local/usercache/user0/filecache/10"),
+ new Path("/local/usercache/user0/filecache/10_tmp")),
+ rsrcTask.getBaseDirs());
+ // nmPrivate token file; the path is null here because localization
+ // failed before it was resolved.
+ FileDeletionTask tokenTask = tasks.get(1);
+ assertNull(tokenTask.getUser());
+ assertNull(tokenTask.getSubDir());
+ assertNull(tokenTask.getBaseDirs());
Review Comment:
These assertions encode a bug. When `getLocalPathForWrite` at the top of
`run()` throws, `nmPrivateCTokensPath` is still null and the finally schedules
`FileDeletionTask(delService, null, null, null)`. `FileDeletionTask.run()`
takes the `user == null && baseDirs == null` branch and calls `lfs.delete(null,
true)`, which NPEs on the DeletionService thread. NPE is not an IOException, so
`deletionTaskFinished()` is skipped and the state store record is never
removed; with recovery enabled it is replayed and NPEs again after every
restart. Pre-existing, but since the test now asserts it as expected, please
fix it in this PR: guard the token file deletion in `LocalizerRunner.run()`
(RLS L1326-L1328) and change this to `times(1)` with only the resource task
asserted.
```java
if (nmPrivateCTokensPath != null) {
FileDeletionTask deletionTask = new FileDeletionTask(delService,
null, nmPrivateCTokensPath, null);
delService.delete(deletionTask);
}
```
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java:
##########
@@ -818,6 +822,76 @@ public void testLocalizerRunnerException() throws
Exception {
}
}
+ @Test
+ @Timeout(value = 10)
+ @SuppressWarnings("unchecked") // mocked generics
+ public void testDownloadingResourcesCleanedUpWhenDispatchFails()
+ throws Exception {
+ Dispatcher dispatcher = mock(Dispatcher.class);
+ EventHandler<Event> eventHandler = mock(EventHandler.class);
+ when(dispatcher.getEventHandler()).thenReturn(eventHandler);
+ // Simulate the localizer thread being interrupted by a container kill:
+ // dispatching the failure event throws instead of completing.
+ Mockito.doThrow(new YarnRuntimeException(new InterruptedException()))
+ .when(eventHandler).handle(isA(ContainerResourceFailedEvent.class));
+
+ ContainerExecutor exec = mock(ContainerExecutor.class);
+ DeletionService delService = mock(DeletionService.class);
+ LocalDirsHandlerService dirsHandlerSpy = spy(new
LocalDirsHandlerService());
+ dirsHandlerSpy.init(conf);
+ // Fail localization so LocalizerRunner.run() takes the error path.
+ Mockito.doThrow(new IOException("Simulated disk failure"))
+ .when(dirsHandlerSpy).getLocalPathForWrite(isA(String.class));
+
+ ResourceLocalizationService rls =
+ new ResourceLocalizationService(dispatcher, exec, delService,
+ dirsHandlerSpy, nmContext, metrics);
+
+ final ApplicationId appId =
+ BuilderUtils.newApplicationId(314159265358979L, 3);
+ final Container c = getMockContainer(appId, 42, "user0");
+ LocalizerRunner runner = rls.new LocalizerRunner(
+ new LocalizerContext("user0", c.getContainerId(), null),
+ c.getContainerId().toString());
+
+ // A resource that was in DOWNLOADING state when the localizer died.
+ LocalizedResource rsrc = mock(LocalizedResource.class);
+ when(rsrc.getLocalPath()).thenReturn(
+ new Path("/local/usercache/user0/filecache/10/foo.jar"));
+ LocalizerResourceRequestEvent scheduledEvent =
+ mock(LocalizerResourceRequestEvent.class);
+ when(scheduledEvent.getResource()).thenReturn(rsrc);
+ runner.scheduled.put(mock(LocalResourceRequest.class), scheduledEvent);
+
+ // Must not propagate the dispatch failure, and must still unlock the
+ // DOWNLOADING resource and schedule the deletion tasks.
+ runner.run();
+
+ // The interrupt status must be restored after the dispatch failure was
+ // swallowed. Thread.interrupted() also clears it for the test thread.
+ assertTrue(Thread.interrupted());
Review Comment:
`run()` sets the flag on the JUnit thread and this assertion is the only
thing that clears it. If `run()` ever throws after the re-interrupt, the flag
leaks into later tests in this class (`Thread.sleep` in
`testLocalizerRunnerException` then fails with a spurious
`InterruptedException`). A flag leaked from an earlier test also makes this
pass vacuously.
```java
assertFalse(Thread.currentThread().isInterrupted());
boolean interrupted;
try {
runner.run();
} finally {
interrupted = Thread.interrupted();
}
assertTrue(interrupted);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
