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 14814adcac0 [FLINK-40446][tests] Make CommonTestUtils.terminateJob
wait for the CANCELED terminal state
14814adcac0 is described below
commit 14814adcac099aa6258cc1f7993760fcd00a5d6d
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 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 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 d11f787aef6..4f42326f9ab 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
@@ -123,7 +123,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 =
@@ -132,7 +132,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)
