This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new 65428744ebe [FLINK-40074][tests] Avoid @TempDir cleanup race in
SinkV2ITCase scaling test
65428744ebe is described below
commit 65428744ebefe69ab659c02b1da5cd4fc48b7735
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 1f2155bc9bf..b63c5dd0ac2 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,10 +54,10 @@ 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 org.slf4j.Logger;
@@ -65,6 +65,7 @@ import org.slf4j.LoggerFactory;
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;
@@ -165,35 +166,47 @@ public class SinkV2ITCase extends AbstractTestBase {
public 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) {
