(flink) branch release-1.19 updated: [FLINK-34571][test] Fix flaky test SortMergeResultPartitionReadSchedulerTest.testOnReadBufferRequestError

[guoweijie]

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

guoweijie pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.19 by this push:
     new 4f7f6a97f79 [FLINK-34571][test] Fix flaky test 
SortMergeResultPartitionReadSchedulerTest.testOnReadBufferRequestError
4f7f6a97f79 is described below

commit 4f7f6a97f799a4a30d1aea57fc0af5d07dc0eeed
Author: Weijie Guo <res...@163.com>
AuthorDate: Fri Mar 8 18:00:09 2024 +0800

    [FLINK-34571][test] Fix flaky test 
SortMergeResultPartitionReadSchedulerTest.testOnReadBufferRequestError
---
 .../partition/SortMergeResultPartitionReadSchedulerTest.java   | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadSchedulerTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadSchedulerTest.java
index c9cb1fa58f4..3d48fc0212d 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadSchedulerTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadSchedulerTest.java
@@ -215,11 +215,19 @@ class SortMergeResultPartitionReadSchedulerTest {
 
     @Test
     void testOnReadBufferRequestError() throws Exception {
+        ManuallyTriggeredScheduledExecutorService schedulerExecutor =
+                new ManuallyTriggeredScheduledExecutorService();
+        readScheduler =
+                new SortMergeResultPartitionReadScheduler(
+                        bufferPool, schedulerExecutor, new Object());
         SortMergeSubpartitionReader subpartitionReader =
                 readScheduler.createSubpartitionReader(
                         new NoOpBufferAvailablityListener(), 0, 
partitionedFile);
-
         bufferPool.destroy();
+        assertThat(schedulerExecutor.numQueuedRunnables()).isEqualTo(1);
+        // we should trigger the scheduled task to handle the buffer request 
error.
+        schedulerExecutor.trigger();
+
         waitUntilReadFinish();
 
         assertThat(subpartitionReader.isReleased()).isTrue();