You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yi...@apache.org on 2022/10/18 03:05:33 UTC
[flink] branch master updated: [FLINK-29641][tests] SortMergeResultPartitionReadSchedulerTest#testCreateSubpartitionReader is not stable
This is an automated email from the ASF dual-hosted git repository.
yingjie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 959fe97beea [FLINK-29641][tests] SortMergeResultPartitionReadSchedulerTest#testCreateSubpartitionReader is not stable
959fe97beea is described below
commit 959fe97beeac15bbac70f4980cc6a3a110b432a2
Author: Weijie Guo <re...@163.com>
AuthorDate: Tue Oct 18 11:05:24 2022 +0800
[FLINK-29641][tests] SortMergeResultPartitionReadSchedulerTest#testCreateSubpartitionReader is not stable
The newly created subpartition reader will trigger the IO thread to read data. If the data reading is completed, the running state will be set to false and the assertion will fail.
This closes #21083.
---
.../partition/SortMergeResultPartitionReadSchedulerTest.java | 9 +++++++++
1 file changed, 9 insertions(+)
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 a865f6c4915..62c76b9f1ec 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
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.concurrent.ManuallyTriggeredScheduledExecutorService;
import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.CompositeBuffer;
@@ -115,7 +116,13 @@ class SortMergeResultPartitionReadSchedulerTest {
}
@Test
+ @Timeout(60)
void testCreateSubpartitionReader() throws Exception {
+ ManuallyTriggeredScheduledExecutorService ioExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ readScheduler =
+ new SortMergeResultPartitionReadScheduler(bufferPool, ioExecutor, new Object());
+
SortMergeSubpartitionReader subpartitionReader =
readScheduler.createSubpartitionReader(
new NoOpBufferAvailablityListener(), 0, partitionedFile);
@@ -123,9 +130,11 @@ class SortMergeResultPartitionReadSchedulerTest {
assertThat(readScheduler.isRunning()).isTrue();
assertThat(readScheduler.getDataFileChannel().isOpen()).isTrue();
assertThat(readScheduler.getIndexFileChannel().isOpen()).isTrue();
+ assertThat(ioExecutor.numQueuedRunnables()).isEqualTo(1);
int numBuffersRead = 0;
while (numBuffersRead < numBuffersPerSubpartition) {
+ ioExecutor.triggerAll();
ResultSubpartition.BufferAndBacklog bufferAndBacklog =
subpartitionReader.getNextBuffer();
if (bufferAndBacklog != null) {