You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/10/06 17:21:36 UTC
[flink] 01/02: [refactor][tests] Remove unnecessary wrapping with
ConsumableNotifyingResultPartitionWriterDecorator
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 91af74e34d22b54100d8718e6e9ecf887911bd6b
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Tue Oct 6 16:21:55 2020 +0200
[refactor][tests] Remove unnecessary wrapping with ConsumableNotifyingResultPartitionWriterDecorator
---
.../runtime/io/network/api/writer/RecordWriterTest.java | 12 ++----------
.../io/benchmark/StreamNetworkBenchmarkEnvironment.java | 13 ++-----------
2 files changed, 4 insertions(+), 21 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java
index c10cd3c..bf6330b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.io.network.api.writer;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -40,7 +39,6 @@ import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
import org.apache.flink.runtime.io.network.partition.NoOpBufferAvailablityListener;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.PipelinedResultPartition;
import org.apache.flink.runtime.io.network.partition.PipelinedSubpartition;
import org.apache.flink.runtime.io.network.partition.PipelinedSubpartitionView;
@@ -54,8 +52,6 @@ import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.apache.flink.runtime.io.network.util.DeserializationUtils;
import org.apache.flink.runtime.operators.shipping.OutputEmitter;
import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
-import org.apache.flink.runtime.taskmanager.ConsumableNotifyingResultPartitionWriterDecorator;
-import org.apache.flink.runtime.taskmanager.NoOpTaskActions;
import org.apache.flink.testutils.serialization.types.SerializationTestType;
import org.apache.flink.testutils.serialization.types.SerializationTestTypeFactory;
import org.apache.flink.testutils.serialization.types.Util;
@@ -305,12 +301,8 @@ public class RecordWriterTest {
.setBufferPoolFactory(() -> localPool)
.build();
resultPartition.setup();
- final ResultPartitionWriter partitionWrapper = new ConsumableNotifyingResultPartitionWriterDecorator(
- new NoOpTaskActions(),
- new JobID(),
- resultPartition,
- new NoOpResultPartitionConsumableNotifier());
- final RecordWriter recordWriter = createRecordWriter(partitionWrapper);
+
+ final RecordWriter<?> recordWriter = createRecordWriter(resultPartition);
try {
// record writer is available because of initial available global pool
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
index f7976a5..d7d7f80 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkBenchmarkEnvironment.java
@@ -30,7 +30,6 @@ import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.netty.NettyConfig;
import org.apache.flink.runtime.io.network.partition.InputChannelTestUtils;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionBuilder;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -42,9 +41,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateFac
import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
-import org.apache.flink.runtime.taskmanager.ConsumableNotifyingResultPartitionWriterDecorator;
import org.apache.flink.runtime.taskmanager.InputGateWithMetrics;
-import org.apache.flink.runtime.taskmanager.NoOpTaskActions;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.ConfigurationParserUtils;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
@@ -188,15 +185,9 @@ public class StreamNetworkBenchmarkEnvironment<T extends IOReadableWritable> {
.setupBufferPoolFactoryFromNettyShuffleEnvironment(senderEnv)
.build();
- ResultPartitionWriter consumableNotifyingPartitionWriter = new ConsumableNotifyingResultPartitionWriterDecorator(
- new NoOpTaskActions(),
- jobId,
- resultPartitionWriter,
- new NoOpResultPartitionConsumableNotifier());
+ resultPartitionWriter.setup();
- consumableNotifyingPartitionWriter.setup();
-
- return consumableNotifyingPartitionWriter;
+ return resultPartitionWriter;
}
private void generatePartitionIds() throws Exception {