This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 7b663bae4c5 [FLINK-40074][tests] Avoid @TempDir cleanup race in
SinkV2ITCase scaling test
7b663bae4c5 is described below
commit 7b663bae4c512f6d32696ded52edbcb62a5b6554
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Jul 6 14:29:09 2026 +0200
[FLINK-40074][tests] Avoid @TempDir cleanup race in SinkV2ITCase scaling
test
writerAndCommitterExecuteInStreamingModeWithScaling injected the checkpoint
directory through JUnit's @TempDir. The class-shared MiniCluster keeps
discarding checkpoint state asynchronously after the finished job returns
its
result, so JUnit's temp directory deletion races with that discard and
intermittently fails with "Failed to delete temp directory" once the job is
done, even though all test assertions pass.
Manage the checkpoint directory explicitly and clean it up quietly with
FileUtils#deleteDirectoryQuietly, which is safe against concurrent deletions
and never lets a cleanup failure mask a real test failure.
Generated-by: Claude Fable 5
---
.../flink/test/streaming/runtime/SinkV2ITCase.java | 67 +++++++++++++---------
1 file changed, 40 insertions(+), 27 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
index 56c136f7bfa..774196d9ffb 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2ITCase.java
@@ -54,15 +54,16 @@ import org.apache.flink.test.junit5.InjectMiniCluster;
import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.testutils.junit.SharedObjectsExtension;
import org.apache.flink.testutils.junit.SharedReference;
+import org.apache.flink.util.FileUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import java.io.File;
import java.io.Serializable;
+import java.nio.file.Files;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -161,35 +162,47 @@ class SinkV2ITCase extends AbstractTestBase {
void writerAndCommitterExecuteInStreamingModeWithScaling(
int initialParallelism,
int scaledParallelism,
- @TempDir File checkpointDir,
@InjectMiniCluster MiniCluster miniCluster,
@InjectClusterClient ClusterClient<?> clusterClient)
throws Exception {
- SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>>
committed =
- SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
- final TrackingCommitter trackingCommitter = new
TrackingCommitter(committed);
- final Configuration config = createConfigForScalingTest(checkpointDir,
initialParallelism);
-
- // first run
- final JobID jobID =
- runStreamingWithScalingTest(
- config,
- initialParallelism,
- trackingCommitter,
- true,
- miniCluster,
- clusterClient);
-
- // second run
- config.set(StateRecoveryOptions.SAVEPOINT_PATH,
getCheckpointPath(miniCluster, jobID));
- config.set(CoreOptions.DEFAULT_PARALLELISM, scaledParallelism);
- runStreamingWithScalingTest(
- config, initialParallelism, trackingCommitter, false,
miniCluster, clusterClient);
-
- assertThat(committed.get())
- .extracting(Committer.CommitRequest::getCommittable)
- .containsExactlyInAnyOrderElementsOf(
- duplicate(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE));
+ // Managed explicitly so cleanup survives the shared MiniCluster's
asynchronous checkpoint
+ // discard after job termination, which would otherwise race JUnit's
@TempDir deletion.
+ final File checkpointDir =
Files.createTempDirectory("SinkV2ITCase").toFile();
+ try {
+ SharedReference<Queue<Committer.CommitRequest<Record<Integer>>>>
committed =
+ SHARED_OBJECTS.add(new ConcurrentLinkedQueue<>());
+ final TrackingCommitter trackingCommitter = new
TrackingCommitter(committed);
+ final Configuration config =
+ createConfigForScalingTest(checkpointDir,
initialParallelism);
+
+ // first run
+ final JobID jobID =
+ runStreamingWithScalingTest(
+ config,
+ initialParallelism,
+ trackingCommitter,
+ true,
+ miniCluster,
+ clusterClient);
+
+ // second run
+ config.set(StateRecoveryOptions.SAVEPOINT_PATH,
getCheckpointPath(miniCluster, jobID));
+ config.set(CoreOptions.DEFAULT_PARALLELISM, scaledParallelism);
+ runStreamingWithScalingTest(
+ config,
+ initialParallelism,
+ trackingCommitter,
+ false,
+ miniCluster,
+ clusterClient);
+
+ assertThat(committed.get())
+ .extracting(Committer.CommitRequest::getCommittable)
+ .containsExactlyInAnyOrderElementsOf(
+
duplicate(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE));
+ } finally {
+ FileUtils.deleteDirectoryQuietly(checkpointDir);
+ }
}
private static List<Record<Integer>> duplicate(List<Record<Integer>>
values) {
