This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4e9d5412d3d [FLINK-40567][tests] Avoid @TempDir cleanup race in
HistoryServerArchiveFetcherTest (#29116)
4e9d5412d3d is described below
commit 4e9d5412d3dcd4776b750e8a94ba6e86d2b409c6
Author: Purushottam Sinha <[email protected]>
AuthorDate: Wed Sep 9 00:13:07 2026 +0530
[FLINK-40567][tests] Avoid @TempDir cleanup race in
HistoryServerArchiveFetcherTest (#29116)
Each test creates a HistoryServerArchiveFetcher that owns two fixed thread
pools which write archives into the @TempDir, but the fetcher was never
closed - tearDown() only closed the storage. Its executor threads could
therefore still be writing when JUnit deleted the temp directory, failing
the run with "Failed to close extension context" /
"Failed to delete temp directory ... DirectoryNotEmptyException: .../jobs".
Track the fetcher in a field and close it in @AfterEach before the storage,
so ExecutorUtils.gracefulShutdown drains both pools before the @TempDir is
removed. This also removes the same latent thread leak in the other tests.
Generated-by: Claude Code (claude-opus-4-8)
---
.../history/HistoryServerArchiveFetcherTest.java | 47 +++++++++++-----------
1 file changed, 24 insertions(+), 23 deletions(-)
diff --git
a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
index 612d156f244..b95449337ec 100644
---
a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
+++
b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java
@@ -71,6 +71,7 @@ class HistoryServerArchiveFetcherTest {
private ArchiveStorage<Object> archiveStorage;
private ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache;
private List<HistoryServerArchiveFetcher.ArchiveEvent> archiveEvents;
+ private HistoryServerArchiveFetcher<?> fetcher;
@Parameters(name = "storageFactory={0}")
private static Collection<ArchiveStorageFactory<?>> storageFactories() {
@@ -95,9 +96,19 @@ class HistoryServerArchiveFetcherTest {
@AfterEach
void tearDown() throws Exception {
archiveEvents.clear();
- if (archiveStorage != null) {
- archiveStorage.close();
- archiveStorage = null;
+ // Shut the fetcher down before deleting the @TempDir: its executor
threads write into
+ // localArchiveRootPath, and a task still running during cleanup fails
the temp-dir
+ // deletion.
+ try {
+ if (fetcher != null) {
+ fetcher.close();
+ }
+ } finally {
+ fetcher = null;
+ if (archiveStorage != null) {
+ archiveStorage.close();
+ archiveStorage = null;
+ }
}
}
@@ -139,8 +150,7 @@ class HistoryServerArchiveFetcherTest {
createJobArchive(remoteArchiveRootPath, jobId, true);
}
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
fetcher.fetchArchives(EAGER);
assertThat(archiveEvents).hasSize(numJobs);
@@ -170,8 +180,7 @@ class HistoryServerArchiveFetcherTest {
// Replace the field so that close() in tearDown() releases the
underlying storage too.
archiveStorage = blockingStorage;
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
fetcher.fetchArchives(LAZY);
@@ -202,8 +211,7 @@ class HistoryServerArchiveFetcherTest {
createJobArchive(remoteArchiveRootPath, jobId, true);
}
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
fetcher.fetchArchives(LAZY);
assertThat(archiveEvents).hasSize(numJobs);
@@ -229,8 +237,7 @@ class HistoryServerArchiveFetcherTest {
Path archivePath = new Path(remoteArchiveRootPath.toURI().toString(),
jobId.toString());
FsJsonArchivist.writeArchivedJsons(archivePath,
Collections.emptyList());
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
assertThatThrownBy(
() -> fetcher.lazyProcessJobArchive(jobId.toString(),
archivePath, false))
@@ -258,8 +265,7 @@ class HistoryServerArchiveFetcherTest {
archiveStorage, "jobs/" + jobIdWithDetail + "/config");
archiveStorage = blockingStorage;
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
fetcher.fetchArchives(LAZY);
@@ -313,8 +319,7 @@ class HistoryServerArchiveFetcherTest {
createJobArchive(remoteArchiveRootPath, job1, true);
createJobArchive(remoteArchiveRootPath, job2, true);
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
fetcher.fetchArchives(EAGER);
Object overviewObject = archiveStorage.getEntry("jobs/overview.json");
@@ -345,8 +350,7 @@ class HistoryServerArchiveFetcherTest {
void testLegacyJobOverviewMigration() throws Exception {
JobID jobId = createLegacyArchive(remoteArchiveRootPath.toPath(),
false);
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
fetcher.fetchArchives(LAZY);
assertThat(archiveEvents).hasSize(1);
@@ -361,8 +365,7 @@ class HistoryServerArchiveFetcherTest {
JobID jobId = JobID.generate();
createJobArchive(remoteArchiveRootPath, jobId, true);
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, true,
archiveStorage);
fetcher.scanArchives(EAGER, false);
assertThat(archiveEvents).isEmpty();
@@ -375,8 +378,7 @@ class HistoryServerArchiveFetcherTest {
JobID jobId = JobID.generate();
Path archivePath = createJobArchive(remoteArchiveRootPath, jobId,
true);
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
archiveStorage);
fetcher.lazyFetchArchiveProactively(jobId.toString(), archivePath);
waitForArchiveLoaded(archiveMetaInfoCache, jobId.toString());
@@ -410,8 +412,7 @@ class HistoryServerArchiveFetcherTest {
archiveStorage, "jobs/" + jobId + "/config");
archiveStorage = blockingStorage;
- HistoryServerArchiveFetcher<?> fetcher =
- createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
+ fetcher = createArchiveFetcher(remoteArchiveRootPath, false,
blockingStorage);
fetcher.fetchArchives(LAZY);
// make sure the async detail task has started
assertThat(blockingStorage.asyncStartLatch.await(10,
TimeUnit.SECONDS)).isTrue();
