(flink) branch release-2.3 updated: [FLINK-39922][runtime/test] Fix the flaky test case AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28426)

[panyuepeng]

This is an automated email from the ASF dual-hosted git repository.

RocMarshal 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 a2d6a26aad7 [FLINK-39922][runtime/test] Fix the flaky test case 
AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28426)
a2d6a26aad7 is described below

commit a2d6a26aad75776afb875f7333ffe9a9a4728aa6
Author: Yuepeng Pan <panyuep...@apache.org>
AuthorDate: Sun Jun 14 10:19:16 2026 +0800

    [FLINK-39922][runtime/test] Fix the flaky test case 
AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28426)
---
 .../operators/AbstractAsyncRunnableStreamOperatorTest.java             | 3 +++
 1 file changed, 3 insertions(+)

diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
index dedf01de245..b073f6cbed1 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
@@ -241,17 +241,20 @@ public class AbstractAsyncRunnableStreamOperatorTest {
                     (SimpleAsyncExecutionController)
                             ((AbstractAsyncRunnableStreamOperator) 
testHarness.getOperator())
                                     .getAsyncExecutionController();
+            CompletableFuture<Void> unblockAsyncRequest = new 
CompletableFuture<>();
             ((AbstractAsyncRunnableStreamOperator<String>) 
testHarness.getOperator())
                     .setAsyncKeyedContextElement(
                             new StreamRecord<>(Tuple2.of(5, "5")), new 
TestKeySelector());
             ((AbstractAsyncRunnableStreamOperator<String>) 
testHarness.getOperator())
                     .asyncProcess(
                             () -> {
+                                unblockAsyncRequest.get(10, TimeUnit.SECONDS);
                                 return null;
                             });
             ((AbstractAsyncRunnableStreamOperator<String>) 
testHarness.getOperator())
                     .postProcessElement();
             
assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1);
+            unblockAsyncRequest.complete(null);
             testHarness.drainAsyncRequests();
             
assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(0);
         }