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 daeebe2bf5c [FLINK-40446][tests] Make CommonTestUtils.terminateJob
wait for the CANCELED terminal state
daeebe2bf5c is described below
commit daeebe2bf5c514fd9fec517d7692707029da83fd
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 | 4 ++--
5 files changed, 15 insertions(+), 21 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 6e2a1231905..44b520ffaf5 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
@@ -291,7 +291,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;
}
@@ -330,7 +330,7 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
externalContext.createSinkDataReader(sinkSettings),
testRecords, semantic);
} finally {
executorService.shutdown();
- killJob(restartJobClient);
+ terminateJob(restartJobClient);
iterator.close();
}
}
@@ -408,7 +408,7 @@ public abstract class SinkTestSuiteBase<T extends
Comparable<T>> {
} finally {
// Clean up
executorService.shutdown();
- killJob(jobClient);
+ terminateJob(jobClient);
}
}
@@ -565,11 +565,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 840f1f6d93a..059a211971d 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
@@ -333,7 +333,7 @@ public abstract class SourceTestSuiteBase<T> {
semantic,
getTestDataSize(testRecordCollections));
} catch (Exception e) {
- killJob(jobClient);
+ terminateJob(jobClient);
throw e;
}
String savepointPath =
@@ -394,7 +394,7 @@ public abstract class SourceTestSuiteBase<T> {
getTestDataSize(newTestRecordCollections));
} finally {
// Clean up
- killJob(restartJobClient);
+ terminateJob(restartJobClient);
iterator.close();
}
}
@@ -469,7 +469,7 @@ public abstract class SourceTestSuiteBase<T> {
} finally {
// Clean up
executorService.shutdown();
- killJob(jobClient);
+ terminateJob(jobClient);
}
}
@@ -622,7 +622,6 @@ public abstract class SourceTestSuiteBase<T> {
// Step 8: Clean up
terminateJob(jobClient);
- waitForJobStatus(jobClient, singletonList(JobStatus.CANCELED));
iterator.close();
}
@@ -767,11 +766,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 6393ad5f200..7c97cfac5f4 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
@@ -104,7 +104,7 @@ class UnalignedCheckpointRescaleSameUpstreamITCase {
CommonTestUtils.waitForCheckpointWithInflightBuffers(
initialJobGraph.getJobID(), miniCluster,
CHECKPOINTS_TO_WAIT);
} finally {
- initialJobClient.cancel().get();
+ CommonTestUtils.terminateJob(initialJobClient);
}
final JobGraph restoredJobGraph = createJobGraph(checkpointPath,
RESTORED_PARALLELISM);
@@ -164,7 +164,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 01ff85fcca8..e443bbcfad4 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
@@ -124,7 +124,7 @@ public class
UnalignedCheckpointRescaleWithMixedExchangesITCase extends TestLogg
String checkpointPath =
CommonTestUtils.waitForCheckpointWithInflightBuffers(
jobClient1.getJobID(), miniCluster);
- jobClient1.cancel().get();
+ CommonTestUtils.terminateJob(jobClient1);
// Step 2: Restore the job with a different parallelism
JobClient jobClient2 =
@@ -133,7 +133,7 @@ public class
UnalignedCheckpointRescaleWithMixedExchangesITCase extends TestLogg
CommonTestUtils.waitForJobStatus(jobClient2,
Collections.singletonList(JobStatus.RUNNING));
CommonTestUtils.waitForAllTaskRunning(miniCluster,
jobClient2.getJobID(), false);
CommonTestUtils.waitForCheckpointWithInflightBuffers(jobClient2.getJobID(),
miniCluster);
- jobClient2.cancel().get();
+ CommonTestUtils.terminateJob(jobClient2);
}
private StreamExecutionEnvironment getUnalignedCheckpointEnv(@Nullable
String recoveryPath)
