This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new 2671b891db4 [FLINK-40074][tests] Avoid @TempDir cleanup race in
SinkV2ITCase scaling test
2671b891db4 is described below
commit 2671b891db4debdf42c55c6e35ad94fa8838bce0
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
(cherry picked from commit 6a99fec07503099fb14dc0e623f67358b91f3203)
---
.../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) {
