From 5fd0b0f076ef8cd697e942ef304c9be17ccfa69d Mon Sep 17 00:00:00 2001 From: Jeremy Schoemaker Date: Tue, 25 Aug 2026 15:44:13 -0500 Subject: [PATCH 1/2] fix(yarn): retain all deletionTasks in rolling log aggregation Fix verified RED->GREEN. YARN-11963 rolling log aggregation leaks local files - single deletionTask overwritten per loop only last container deleted --- .../logaggregation/AppLogAggregatorImpl.java | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java index ce6397e3904817..73fa030783916f 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java @@ -332,7 +332,7 @@ private void uploadLogsForContainers(boolean appFinished) LOG.debug("Cycle #{} of log aggregator", logAggregationTimes); String diagnosticMessage = ""; boolean logAggregationSucceedInThisCycle = true; - DeletionTask deletionTask = null; + List deletionTasks = new ArrayList<>(); try { try { logAggregationFileController.initializeWriter(logControllerContext); @@ -374,8 +374,9 @@ private void uploadLogsForContainers(boolean appFinished) } } } - deletionTask = new FileDeletionTask(delService, this.userUgi.getShortUserName(), null, - uploadedFilePathsInThisCycleList); + deletionTasks.add(new FileDeletionTask(delService, + this.userUgi.getShortUserName(), null, + uploadedFilePathsInThisCycleList)); } } @@ -411,8 +412,10 @@ private void uploadLogsForContainers(boolean appFinished) logAggregationSucceedInThisCycle = false; exc = e; } - if (logAggregationSucceedInThisCycle && deletionTask != null) { - delService.delete(deletionTask); + if (logAggregationSucceedInThisCycle) { + for (DeletionTask deletionTask : deletionTasks) { + delService.delete(deletionTask); + } } if (diagnosticMessage != null && !diagnosticMessage.isEmpty()) { LOG.debug("Sending log aggregation report along with the " + From 040fa4436aab236bcb4d19f78a2587fc324828ad Mon Sep 17 00:00:00 2001 From: Jeremy Schoemaker Date: Sun, 30 Aug 2026 03:20:27 -0500 Subject: [PATCH 2/2] test(yarn): cover multi-container deletion in rolling log aggregation Add a regression test for YARN-11963. TestAppLogAggregatorImpl only ever exercised a single container, so both the previous single-variable implementation and the fix pass the existing suite. The new test starts two containers with distinct log files in one aggregation cycle and asserts that the uploaded files of both are scheduled for deletion. Verified RED->GREEN: against the single-variable implementation it fails with only the last container's files scheduled. --- .../TestAppLogAggregatorImpl.java | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestAppLogAggregatorImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestAppLogAggregatorImpl.java index 9474edc890cb17..a95b96bfad3cbb 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestAppLogAggregatorImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestAppLogAggregatorImpl.java @@ -69,6 +69,7 @@ import java.util.Map; import java.util.Set; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; @@ -180,6 +181,112 @@ public void testAggregatorWhenAllFilesOlderThanRetentionShouldUploadNone() logFiles, new HashSet()); } + /** + * Regression test for YARN-11963. A single aggregation cycle that covers + * more than one container must schedule the deletion of the uploaded log + * files of every container in the cycle, not just those of the container + * that happened to be processed last. + */ + @Test + public void testAggregatorDeletesUploadedLogsOfEveryContainerInACycle() + throws Exception { + final ApplicationId applicationId = + ApplicationId.newInstance(System.currentTimeMillis(), 0); + final ApplicationAttemptId attemptId = + ApplicationAttemptId.newInstance(applicationId, 0); + final ContainerId firstContainerId = + ContainerId.newContainerId(attemptId, 1); + final ContainerId secondContainerId = + ContainerId.newContainerId(attemptId, 2); + + // create artificial log files for two distinct containers + final File appLogDir = new File(LOCAL_LOG_DIR, applicationId.toString()); + final File firstContainerLogDir = + new File(appLogDir, firstContainerId.toString()); + final File secondContainerLogDir = + new File(appLogDir, secondContainerId.toString()); + firstContainerLogDir.mkdirs(); + secondContainerLogDir.mkdirs(); + final Set firstContainerLogFiles = + createContainerLogFiles(firstContainerLogDir, 3); + final Set secondContainerLogFiles = + createContainerLogFiles(secondContainerLogDir, 3); + + final Set filesExpected2Delete = new HashSet<>(); + for (File file: firstContainerLogFiles) { + filesExpected2Delete.add(file.getAbsolutePath()); + } + for (File file: secondContainerLogFiles) { + filesExpected2Delete.add(file.getAbsolutePath()); + } + + // Collect the paths of every deletion task scheduled during the cycle. + // Checking only the first invocation of delete() would not catch the + // bug: the buggy implementation kept a single DeletionTask reference + // that each container overwrote, so it still issued one delete() call - + // just with the wrong, and incomplete, set of paths. + final Set filesActually2Delete = new HashSet<>(); + final DeletionService deletionService = + createDeletionServiceCapturingAllDeletions(filesActually2Delete); + + final YarnConfiguration config = new YarnConfiguration(); + config.setLong(YarnConfiguration.LOG_AGGREGATION_RETAIN_SECONDS, 10000); + + LogAggregationTFileController format = + spy(new LogAggregationTFileController()); + format.initialize(config, "TFile"); + + final Context context = createContext(config); + final AppLogAggregatorInTest appLogAggregator = + createAppLogAggregator(applicationId, LOCAL_LOG_DIR.getAbsolutePath(), + config, context, -1, deletionService, format); + + appLogAggregator.startContainerLogAggregation( + new ContainerLogContext(firstContainerId, ContainerType.TASK, 0)); + appLogAggregator.startContainerLogAggregation( + new ContainerLogContext(secondContainerId, ContainerType.TASK, 0)); + // set app finished flag first + appLogAggregator.finishLogAggregation(); + appLogAggregator.run(); + + // Assert containment rather than set equality: on top of the per + // container deletion tasks, the aggregator also schedules the + // application log directory itself for cleanup once the application + // has finished. + for (String file: filesExpected2Delete) { + assertTrue(filesActually2Delete.contains(file), + "Expected the uploaded log file " + file + " to be scheduled for " + + "deletion, but the scheduled paths were " + + filesActually2Delete); + } + } + + /** + * Create a DeletionService that records the paths of every FileDeletionTask + * handed to its delete method, across all invocations, into the given set. + * @param deletedPaths the set collecting the absolute paths seen + * @return the recording DeletionService mock + */ + private static DeletionService createDeletionServiceCapturingAllDeletions( + final Set deletedPaths) { + DeletionService recordingDeletionService = mock(DeletionService.class); + doAnswer(new Answer() { + @Override + public Void answer(InvocationOnMock invocationOnMock) { + for (Object taskArgument: invocationOnMock.getArguments()) { + FileDeletionTask task = (FileDeletionTask) taskArgument; + for (Path path: task.getBaseDirs()) { + deletedPaths.add( + new File(path.toUri().getRawPath()).getAbsolutePath()); + } + } + return null; + } + }).when(recordingDeletionService).delete(any(FileDeletionTask.class)); + + return recordingDeletionService; + } + /** * Create the given number of log files under the container log directory. * @param containerLogDir the directory to create container log files