This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new c47b8eefd2d [FLINK-40074][tests] Avoid @TempDir cleanup race in
SinkV2ITCase scaling test
c47b8eefd2d is described below
commit c47b8eefd2d23f688fc8ad4d0862c97d42f616f5
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Jul 6 14:18:25 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 | 51 +++++++++++++---------
1 file changed, 30 insertions(+), 21 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 92690b73f43..81c709bf671 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
@@ -52,15 +52,16 @@ import org.apache.flink.streaming.util.FiniteTestSource;
import org.apache.flink.test.junit5.InjectClusterClient;
import org.apache.flink.test.junit5.InjectMiniCluster;
import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.util.FileUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-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.Collections;
import java.util.List;
@@ -227,29 +228,37 @@ public class SinkV2ITCase extends AbstractTestBase {
public void writerAndCommitterExecuteInStreamingModeWithScaling(
int initialParallelism,
int scaledParallelism,
- @TempDir File checkpointDir,
@InjectMiniCluster MiniCluster miniCluster,
@InjectClusterClient ClusterClient<?> clusterClient)
throws Exception {
- final DefaultCommitter committer =
- new DefaultCommitter(
- (Supplier<Queue<Committer.CommitRequest<String>>> &
Serializable)
- () -> COMMIT_QUEUE);
- final Configuration config = createConfigForScalingTest(checkpointDir,
initialParallelism);
-
- // first run
- final JobID jobID = runStreamingWithScalingTest(config, true,
committer, clusterClient);
-
- // second run
- config.set(StateRecoveryOptions.SAVEPOINT_PATH,
getCheckpointPath(miniCluster, jobID));
- config.set(CoreOptions.DEFAULT_PARALLELISM, scaledParallelism);
- runStreamingWithScalingTest(config, false, committer, clusterClient);
-
- assertThat(
- COMMIT_QUEUE.stream()
- .map(Committer.CommitRequest::getCommittable)
- .collect(Collectors.toList()),
-
containsInAnyOrder(duplicate(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE).toArray()));
+ // 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 {
+ final DefaultCommitter committer =
+ new DefaultCommitter(
+ (Supplier<Queue<Committer.CommitRequest<String>>>
& Serializable)
+ () -> COMMIT_QUEUE);
+ final Configuration config =
+ createConfigForScalingTest(checkpointDir,
initialParallelism);
+
+ // first run
+ final JobID jobID = runStreamingWithScalingTest(config, true,
committer, clusterClient);
+
+ // second run
+ config.set(StateRecoveryOptions.SAVEPOINT_PATH,
getCheckpointPath(miniCluster, jobID));
+ config.set(CoreOptions.DEFAULT_PARALLELISM, scaledParallelism);
+ runStreamingWithScalingTest(config, false, committer,
clusterClient);
+
+ assertThat(
+ COMMIT_QUEUE.stream()
+ .map(Committer.CommitRequest::getCommittable)
+ .collect(Collectors.toList()),
+ containsInAnyOrder(
+
duplicate(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE).toArray()));
+ } finally {
+ FileUtils.deleteDirectoryQuietly(checkpointDir);
+ }
}
private static List<String> duplicate(List<String> values) {
