YARN-11856. DOWNLOADING resources unlock and cleanup is interrupted when killing a container that is localizing - #8691
YARN-11856. DOWNLOADING resources unlock and cleanup is interrupted when killing a container that is localizing#8691eubnara wants to merge 2 commits into
Conversation
…hen killing a container that is localizing. Contributed by zheng-weihao and Yubi Lee.
|
💔 -1 overall
This message was automatically generated. |
There was a problem hiding this comment.
Pull request overview
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Fixes NodeManager localization cleanup when a container is killed during localization and event dispatch fails due to interruption, ensuring resources aren’t left stuck in DOWNLOADING and cleanup work still runs.
Changes:
- Wrap
ContainerResourceFailedEventdispatch inLocalizerRunner.run()with atry/catchso cleanup logic always executes. - Add a unit test that simulates a dispatch failure (
YarnRuntimeException(InterruptedException)) and asserts resources are unlocked and deletions are scheduled.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
.../ResourceLocalizationService.java |
Ensures cleanup runs even if dispatching ContainerResourceFailedEvent throws during interruption. |
.../TestResourceLocalizationService.java |
Adds regression test covering dispatch-failure cleanup behavior for DOWNLOADING resources. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
…evel logging, stronger test assertions. - Restore the thread interrupt status when the swallowed dispatch failure was caused by an InterruptedException. - Log the dispatch failure at WARN with parameterized logging. - Assert the exact FileDeletionTasks (localization dir + _tmp dir, nmPrivate token file) and the restored interrupt status in the test.
|
Addressed the Copilot review comments in 8600dbb: restored the interrupt status when the swallowed failure was caused by an |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
@slfan1989 @cnauroth Could you take a look when you have a chance? This continues #7893 (YARN-11856): when a container is killed while localizing, dispatching ContainerResourceFailedEvent can throw (InterruptedException from the dispatcher), which escapes LocalizerRunner before the DOWNLOADING resources are unlocked/cleaned up, leaving other containers waiting on the same private resource stuck in LOCALIZING. This PR catches the dispatch failure (restoring the interrupt flag) so the cleanup always runs, plus a test reproducing the kill-during-localization case. Thanks! |
slfan1989
left a comment
slfan1989
left a comment
There was a problem hiding this comment.
Thanks for addressing the previous review comments.
I reviewed the latest revision and the fix looks good to me. Catching the dispatch failure ensures that the remaining cleanup always runs, while restoring the interrupt status preserves the LocalizerRunner thread's interrupt semantics.
The updated test now verifies the exact deletion tasks for the localization directory, the _tmp directory, and the nmPrivate token file, as well as the restored interrupt status.
The current ASF license failure is unrelated to this PR and comes from the vendored JSON.java file in hadoop-hdfs-rbf.
No blocking issues from my side. +1.
pan3793
left a comment
pan3793
left a comment
There was a problem hiding this comment.
The description says every kill-during-localization hits this path and that this is the YARN-side trigger of the DFS socket leak. That overstates it. Under DefaultContainerExecutor the runner thread is usually blocked in ContainerLocalizer.localizeFiles at cs.poll, where the interrupt is swallowed by catch (InterruptedException e) { return; }; startLocalizer then returns normally, exception is null, and the dispatch is never reached. The flag survives to the dispatch only when the interrupt lands inside the heartbeat RPC (ipc Client re-sets it) and awaitTermination in runLocalization's finally does not consume it. Under LinuxContainerExecutor, Shell.runCommand turns the interrupt into InterruptedIOException without re-interrupting, so the dispatch normally succeeds. Nothing here cancels the in-flight FSDownload (exec.shutdown(), not shutdownNow()), so the NM-side socket leak is not affected by this change. Please scope the description to the DOWNLOADING unlock and cleanup fix.
Otherwise the shape of the fix is right: the dispatch must not abort the cleanup, and dropping the event on interrupt is correct (see the inline comment on the catch block).
Two more items outside the diff. Both are pre-existing, but this change makes them reachable on the kill path, so please handle them in this PR.
DeletionService.delete during NM stop (ResourceLocalizationService.java L1324 / L1328)
DeletionService is added before ContainerManager in NodeManager, so it stops after it. LocalizerTracker.serviceStop interrupts runners without joining, and a DCE runner can spend up to 10s in ContainerLocalizer's awaitTermination before reaching the finally, by which time DeletionService.serviceStop has called sched.shutdown(). DeletionService.delete has no guard, so schedule() throws RejectedExecutionException out of the finally and the token file deletion is skipped; with recovery enabled nmPrivate is not wiped on restart. The PR text says the cleanup always runs, so please close this as well. recordDeletionTaskInStateStore runs before schedule, so catching lets recovery replay the task:
// DeletionService.java
public void delete(DeletionTask deletionTask) {
if (debugDelay != -1) {
LOG.debug("Scheduling DeletionTask (delay {}) : {}", debugDelay,
deletionTask);
recordDeletionTaskInStateStore(deletionTask);
try {
sched.schedule(deletionTask, debugDelay, TimeUnit.SECONDS);
} catch (RejectedExecutionException e) {
// NM is stopping. With recovery enabled the task is replayed on
// restart.
LOG.warn("DeletionService is stopped, skip {}", deletionTask);
}
}
}Stacked in-progress paths on retry (LocalResourcesTrackerImpl.getPathForLocalization, L462)
With this PR, a killed container's runner unlocks a resource that is still DOWNLOADING when another container holds a reference (refCount > 0 after RELEASE). The waiting container's runner then tryAcquire()s it and calls getPathForLocalization for a resource that already has a local path from the abandoned attempt: getRelativePathForLocalization increments the directory count again, inProgressLocalResourcesMap.put overwrites the old path, and a second started/ record is written. Nothing removes the old started/ record, because removeResource and finishResourceLocalization only use the current getLocalPath(). On the next restart with recovery enabled, recoverTrackerResources replays the new completed/ record, then hits the stale started/ record and calls tracker.remove() on the resource, deleting the good copy. The resource is re-downloaded after every restart and the directory count leaks.
Sketch, at the top of getPathForLocalization before inProgressLocalResourcesMap.put:
LocalizedResource rsrc = localrsrc.get(req);
if (rsrc != null && rsrc.getLocalPath() != null) {
// A previous attempt was abandoned (container killed while
// downloading). Drop its bookkeeping before allocating a new path,
// otherwise the stale started/ record removes the good copy on recovery.
decrementFileCountForLocalCacheDirectory(req, rsrc);
try {
stateStore.removeLocalizedResource(user, appId, rsrc.getLocalPath());
} catch (IOException e) {
LOG.error("Unable to remove stale localization record for " + rsrc, e);
}
}A test for the shared-resource kill case (two containers, one killed while downloading, then restart and recover) would cover both this and the runner change.
| } 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(); | ||
| } | ||
| } |
There was a problem hiding this 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.
| } 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(); | |
| } | |
| } | |
| } 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); | |
| } | |
| } |
| } | ||
| } | ||
| List<Path> paths = new ArrayList<Path>(); | ||
| for (LocalizerResourceRequestEvent event : scheduled.values()) { |
There was a problem hiding this 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:
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".
| @SuppressWarnings("unchecked") // mocked generics | ||
| public void testDownloadingResourcesCleanedUpWhenDispatchFails() | ||
| throws Exception { | ||
| Dispatcher dispatcher = mock(Dispatcher.class); |
There was a problem hiding this 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.
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.
| LocalDirsHandlerService dirsHandlerSpy = spy(new LocalDirsHandlerService()); | ||
| dirsHandlerSpy.init(conf); |
There was a problem hiding this 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.
|
|
||
| // The interrupt status must be restored after the dispatch failure was | ||
| // swallowed. Thread.interrupted() also clears it for the test thread. | ||
| assertTrue(Thread.interrupted()); |
There was a problem hiding this 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.
assertFalse(Thread.currentThread().isInterrupted());
boolean interrupted;
try {
runner.run();
} finally {
interrupted = Thread.interrupted();
}
assertTrue(interrupted);| // 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()); |
There was a problem hiding this 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.
if (nmPrivateCTokensPath != null) {
FileDeletionTask deletionTask = new FileDeletionTask(delService,
null, nmPrivateCTokensPath, null);
delService.delete(deletionTask);
}
Description of PR
When a container is killed while it is localizing, the
LocalizerRunnerthread is interrupted. In the
finallyblock ofLocalizerRunner.run(),dispatching the
ContainerResourceFailedEventthen throwsYarnRuntimeException(InterruptedException), which skips everything afterit: resources left in DOWNLOADING state are never unlocked, and the
deletion tasks for the localization dirs, the
_tmpdownload dirs and thenmPrivate token file are never scheduled.
This PR revives the fix from #7893 by @zheng-weihao (stale-closed after
100 days of inactivity), unchanged: the event dispatch is wrapped in
try/catch so the cleanup below it always runs. The remaining review
comment on #7893 (dropping
FSErrorfrom the catch list) had already beenaddressed in its final revision. Credit to the original author is kept in
the commit message.
What this PR adds on top of #7893 is a unit test, which was the other
blocker ("no new or modified tests").
We hit this in production on a ~600 NodeManager cluster running
DefaultContainerExecutor: because DCE runs the localizer inside the NM
JVM, every kill-during-localization also leaked DFS block reader sockets
into the long-lived NM process, leaving DataNodes with FIN_WAIT1
connections whose send queues never drain. The HDFS side of that leak is
tracked separately in HDFS-17965 (#8690); this issue is the YARN-side
trigger and also leaves resources stuck in DOWNLOADING state regardless of
the container executor in use.
How was this patch tested?
New unit test
TestResourceLocalizationService#testDownloadingResourcesCleanedUpWhenDispatchFails:a mocked dispatcher throws
YarnRuntimeException(InterruptedException)when the
ContainerResourceFailedEventis dispatched, simulating thekill-during-localization interrupt. Without the fix,
run()propagatesthe exception and the DOWNLOADING resource is never unlocked (test fails).
With the fix, the resource is unlocked and the deletion tasks are
scheduled (test passes).
For code changes:
declared according to the connector-specific documentation? Note: Automated CI
testing doesn't cover all cases so manual testing with cloud storage is still
required.
LICENSE,LICENSE-binary,NOTICE-binaryfiles?AI Tooling
If an AI tool was used:
where is the name of the AI tool used.
Contains content generated by Claude Code (Anthropic Claude): the
new unit test. The fix itself is unchanged from YARN-11856. DOWNLOADING resources unlock and cleanup is interrupted w… #7893. All content
was human-reviewed and complies with the ASF Generative Tooling
Guidance (https://www.apache.org/legal/generative-tooling.html).
https://www.apache.org/legal/generative-tooling.html