[FLINK-34571][test] Fix flaky test SortMergeResultPartitionReadSchedulerTest.testOnReadBufferRequestError

Author: reswqa

Diff for: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionReadSchedulerTest.java

@@ -215,11 +215,19 @@ void testOnDataReadError() throws Exception {
 
     @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();