Copilot commented on code in PR #8691:
URL: https://github.com/apache/hadoop/pull/8691#discussion_r3812197987
##########
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,56 @@ 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();
+
+ verify(rsrc).unlock();
+ verify(delService, Mockito.atLeastOnce())
+ .delete(isA(FileDeletionTask.class));
Review Comment:
This assertion only verifies that *some* `FileDeletionTask` was scheduled,
but the PR description says cleanup should cover localization dirs, `_tmp`
download dirs, and the nmPrivate token file. To make the regression test
stronger (and reduce the chance of false positives), consider capturing the
submitted `FileDeletionTask` arguments and asserting the expected paths/tasks
are present (and optionally the expected count).
##########
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,12 @@ 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.info("Failed to send container resource failed event for " +
cId.toString(), e);
Review Comment:
Catching and swallowing an interrupt-related failure here can accidentally
clear/lose the thread interrupt semantics for the remainder of `run()` and any
callers (depending on where the interrupt was thrown/handled). Consider
detecting `InterruptedException` (either `e` itself or `e.getCause()`) and
restoring the interrupt status via `Thread.currentThread().interrupt()` before
proceeding with cleanup/logging.
##########
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,12 @@ 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.info("Failed to send container resource failed event for " +
cId.toString(), e);
Review Comment:
Failure to dispatch a container resource failure event is operationally
significant (it can mask the original failure/abort signal for the container),
so `INFO` may be too low. Consider logging at `WARN` and using parameterized
logging (and avoid the redundant `.toString()`) to keep logs consistent and
cheaper under load.
--
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]
