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 257e5047c2c [FLINK-40446][tests] Make CommonTestUtils.terminateJob
wait for the CANCELED terminal state
257e5047c2c is described below
commit 257e5047c2cfa0b320fb672f74a32bd9df7ed063
Author: Rui Fan <[email protected]>
AuthorDate: Thu Aug 20 14:26:47 2026 +0200
[FLINK-40446][tests] Make CommonTestUtils.terminateJob wait for the
CANCELED terminal state
JobClient#cancel().get() only acknowledges that cancellation was initiated,
not
that the job has reached a terminal state. Tests that cancel a job and then
let
JUnit delete its checkpoint directory (a @TempDir) can therefore race with
the
source subtasks' async snapshot of the last completed checkpoint, which
keeps
writing SourceReaderState files into the file-merging taskowned directory
after
cancel() returns. This shows up as a flaky
UnalignedCheckpointRescaleWithMixedExchangesITCase teardown failure:
"Failed to delete temp directory ... DirectoryNotEmptyException".
Make CommonTestUtils.terminateJob wait for the CANCELED terminal state after
cancel(), and route the UC rescale ITCases and the connector test suites
(which previously duplicated this wait in a local killJob helper) through
it.
(cherry picked from commit 9c1015353f8735bf28b76d01db544514fab60c87)
---
.../org/apache/flink/runtime/testutils/CommonTestUtils.java | 5 +++++
.../connector/testframe/testsuites/SinkTestSuiteBase.java | 11 +++--------
.../connector/testframe/testsuites/SourceTestSuiteBase.java | 12 +++---------
.../UnalignedCheckpointRescaleSameUpstreamITCase.java | 4 ++--
.../UnalignedCheckpointRescaleWithMixedExchangesITCase.java | 6 +++---
5 files changed, 16 insertions(+), 22 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
index 9006f5b4662..67812b27878 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
@@ -49,6 +49,7 @@ import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -300,7 +301,11 @@ public class CommonTestUtils {
}
public static void terminateJob(JobClient client) throws Exception {
+ // cancel() only acknowledges that cancellation was initiated; wait
for the terminal state
+ // so callers can rely on the job having fully stopped (e.g. before
deleting checkpoint dirs
+ // or asserting on final state) rather than each caller re-adding the
wait themselves.
client.cancel().get();
+ waitForJobStatus(client,
Collections.singletonList(JobStatus.CANCELED));
}
public static void waitForSubtasksToFinish(
diff --git
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
index 16e1854cd27..c0facf648e3 100644
---
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
+++
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
@@ -289,7 +289,7 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
waitForJobStatus(jobClient,
Collections.singletonList(JobStatus.FINISHED));
} catch (Exception e) {
executorService.shutdown();
- killJob(jobClient);
+ terminateJob(jobClient);
throw e;
}
@@ -328,7 +328,7 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
externalContext.createSinkDataReader(sinkSettings),
testRecords, semantic);
} finally {
executorService.shutdown();
- killJob(restartJobClient);
+ terminateJob(restartJobClient);
iterator.close();
}
}
@@ -406,7 +406,7 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
} finally {
// Clean up
executorService.shutdown();
- killJob(jobClient);
+ terminateJob(jobClient);
}
}
@@ -563,11 +563,6 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
return
TestingSinkSettings.builder().setCheckpointingMode(checkpointingMode).build();
}
- private void killJob(JobClient jobClient) throws Exception {
- terminateJob(jobClient);
- waitForJobStatus(jobClient,
Collections.singletonList(JobStatus.CANCELED));
- }
-
private DataStreamSink<T> tryCreateSink(
DataStream<T> dataStream,
DataStreamSinkExternalContext<T> context,
diff --git
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
index 95feb5e3f5c..2cd90a08e96 100644
---
a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
+++
b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
@@ -332,7 +332,7 @@ public abstract class SourceTestSuiteBase<T> {
semantic,
getTestDataSize(testRecordCollections));
} catch (Exception e) {
- killJob(jobClient);
+ terminateJob(jobClient);
throw e;
}
String savepointPath =
@@ -393,7 +393,7 @@ public abstract class SourceTestSuiteBase<T> {
getTestDataSize(newTestRecordCollections));
} finally {
// Clean up
- killJob(restartJobClient);
+ terminateJob(restartJobClient);
iterator.close();
}
}
@@ -468,7 +468,7 @@ public abstract class SourceTestSuiteBase<T> {
} finally {
// Clean up
executorService.shutdown();
- killJob(jobClient);
+ terminateJob(jobClient);
}
}
@@ -621,7 +621,6 @@ public abstract class SourceTestSuiteBase<T> {
// Step 8: Clean up
terminateJob(jobClient);
- waitForJobStatus(jobClient, singletonList(JobStatus.CANCELED));
iterator.close();
}
@@ -768,11 +767,6 @@ public abstract class SourceTestSuiteBase<T> {
return Precision.equals(allRecordSize, sumNumRecordsIn);
}
- private void killJob(JobClient jobClient) throws Exception {
- terminateJob(jobClient);
- waitForJobStatus(jobClient, singletonList(JobStatus.CANCELED));
- }
-
/** Builder class for constructing {@link CollectResultIterator} of
collect sink. */
protected static class CollectIteratorBuilder<T> {
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleSameUpstreamITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleSameUpstreamITCase.java
index 15ef9366ff3..3f4ba4b9c73 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleSameUpstreamITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleSameUpstreamITCase.java
@@ -111,7 +111,7 @@ class UnalignedCheckpointRescaleSameUpstreamITCase {
CommonTestUtils.waitForCheckpointWithInflightBuffers(
initialJobGraph.getJobID(), miniCluster,
CHECKPOINTS_TO_WAIT);
} finally {
- initialJobClient.cancel().get();
+ CommonTestUtils.terminateJob(initialJobClient);
}
final JobGraph restoredJobGraph =
@@ -177,7 +177,7 @@ class UnalignedCheckpointRescaleSameUpstreamITCase {
private static void cancelIfRunning(JobClient jobClient) throws Exception {
if (jobClient.getJobStatus().get() != JobStatus.FAILED) {
- jobClient.cancel().get();
+ CommonTestUtils.terminateJob(jobClient);
}
}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleWithMixedExchangesITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleWithMixedExchangesITCase.java
index d9061ae3f05..de05dd8d34c 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleWithMixedExchangesITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleWithMixedExchangesITCase.java
@@ -131,7 +131,7 @@ class UnalignedCheckpointRescaleWithMixedExchangesITCase {
String checkpointPath1 =
CommonTestUtils.waitForCheckpointWithInflightBuffers(
jobClient1.getJobID(), miniCluster);
- jobClient1.cancel().get();
+ CommonTestUtils.terminateJob(jobClient1);
LOG.info("First checkpoint path: {}", checkpointPath1);
// Step 2: Restore the job with a different parallelism
@@ -143,7 +143,7 @@ class UnalignedCheckpointRescaleWithMixedExchangesITCase {
String checkpointPath2 =
CommonTestUtils.waitForCheckpointWithInflightBuffers(
jobClient2.getJobID(), miniCluster);
- jobClient2.cancel().get();
+ CommonTestUtils.terminateJob(jobClient2);
LOG.info("Second checkpoint path: {}", checkpointPath2);
// Step 3: Restore from Step 2's checkpoint with random parallelism.
This validates
@@ -155,7 +155,7 @@ class UnalignedCheckpointRescaleWithMixedExchangesITCase {
CommonTestUtils.waitForAllTaskRunning(miniCluster,
jobClient3.getJobID(), false);
// Wait for at least one checkpoint to verify the recovery was
successful
CommonTestUtils.waitForCheckpointWithInflightBuffers(jobClient3.getJobID(),
miniCluster);
- jobClient3.cancel().get();
+ CommonTestUtils.terminateJob(jobClient3);
}
private StreamExecutionEnvironment getUnalignedCheckpointEnv(@Nullable
String recoveryPath)
