This is an automated email from the ASF dual-hosted git repository.
1996fanrui 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 9c1015353f8 [FLINK-40446][tests] Make CommonTestUtils.terminateJob
wait for the CANCELED terminal state
9c1015353f8 is described below
commit 9c1015353f8735bf28b76d01db544514fab60c87
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.
---
.../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 557c49395ca..15ddb7693fd 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 9664e109e84..da9adfb25d1 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
@@ -132,7 +132,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
@@ -144,7 +144,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
@@ -156,7 +156,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)
