You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/04/04 13:31:08 UTC
[10/19] flink git commit: [FLINK-8704][tests] Port
PartialConsumerPipelinedResultTest
[FLINK-8704][tests] Port PartialConsumerPipelinedResultTest
This closes #5695.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fc1244eb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fc1244eb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fc1244eb
Branch: refs/heads/release-1.5
Commit: fc1244eb9e8197bb0437aaa210f0ecaaa42af98c
Parents: 40b9a63
Author: zentol <ch...@apache.org>
Authored: Mon Mar 12 13:09:44 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Wed Apr 4 14:26:46 2018 +0200
----------------------------------------------------------------------
...LegacyPartialConsumePipelinedResultTest.java | 150 +++++++++++++++++++
.../PartialConsumePipelinedResultTest.java | 33 ++--
2 files changed, 171 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fc1244eb/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/LegacyPartialConsumePipelinedResultTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/LegacyPartialConsumePipelinedResultTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/LegacyPartialConsumePipelinedResultTest.java
new file mode 100644
index 0000000..aecab75
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/LegacyPartialConsumePipelinedResultTest.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.io.network.partition;
+
+import org.apache.flink.configuration.AkkaOptions;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.runtime.jobgraph.DistributionPattern;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.testingUtils.TestingCluster;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class LegacyPartialConsumePipelinedResultTest extends TestLogger {
+
+ // Test configuration
+ private final static int NUMBER_OF_TMS = 1;
+ private final static int NUMBER_OF_SLOTS_PER_TM = 1;
+ private final static int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
+
+ private final static int NUMBER_OF_NETWORK_BUFFERS = 128;
+
+ private static TestingCluster flink;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ final Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUMBER_OF_TMS);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUMBER_OF_SLOTS_PER_TM);
+ config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+ config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
+
+ flink = new TestingCluster(config, true);
+
+ flink.start();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ flink.stop();
+ }
+
+ /**
+ * Tests a fix for FLINK-1930.
+ *
+ * <p> When consuming a pipelined result only partially, is is possible that local channels
+ * release the buffer pool, which is associated with the result partition, too early. If the
+ * producer is still producing data when this happens, it runs into an IllegalStateException,
+ * because of the destroyed buffer pool.
+ *
+ * @see <a href="https://issues.apache.org/jira/browse/FLINK-1930">FLINK-1930</a>
+ */
+ @Test
+ public void testPartialConsumePipelinedResultReceiver() throws Exception {
+ final JobVertex sender = new JobVertex("Sender");
+ sender.setInvokableClass(SlowBufferSender.class);
+ sender.setParallelism(PARALLELISM);
+
+ final JobVertex receiver = new JobVertex("Receiver");
+ receiver.setInvokableClass(SingleBufferReceiver.class);
+ receiver.setParallelism(PARALLELISM);
+
+ // The partition needs to be pipelined, otherwise the original issue does not occur, because
+ // the sender and receiver are not online at the same time.
+ receiver.connectNewDataSetAsInput(
+ sender, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
+
+ final JobGraph jobGraph = new JobGraph("Partial Consume of Pipelined Result", sender, receiver);
+
+ final SlotSharingGroup slotSharingGroup = new SlotSharingGroup(
+ sender.getID(), receiver.getID());
+
+ sender.setSlotSharingGroup(slotSharingGroup);
+ receiver.setSlotSharingGroup(slotSharingGroup);
+
+ flink.submitJobAndWait(jobGraph, false, TestingUtils.TESTING_DURATION());
+ }
+
+ // ---------------------------------------------------------------------------------------------
+
+ /**
+ * Sends a fixed number of buffers and sleeps in-between sends.
+ */
+ public static class SlowBufferSender extends AbstractInvokable {
+
+ public SlowBufferSender(Environment environment) {
+ super(environment);
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ final ResultPartitionWriter writer = getEnvironment().getWriter(0);
+
+ for (int i = 0; i < 8; i++) {
+ final BufferBuilder bufferBuilder = writer.getBufferProvider().requestBufferBuilderBlocking();
+ writer.addBufferConsumer(bufferBuilder.createBufferConsumer(), 0);
+ Thread.sleep(50);
+ bufferBuilder.finish();
+ }
+ }
+ }
+
+ /**
+ * Reads a single buffer and recycles it.
+ */
+ public static class SingleBufferReceiver extends AbstractInvokable {
+
+ public SingleBufferReceiver(Environment environment) {
+ super(environment);
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ InputGate gate = getEnvironment().getInputGate(0);
+ Buffer buffer = gate.getNextBufferOrEvent().orElseThrow(IllegalStateException::new).getBuffer();
+ if (buffer != null) {
+ buffer.recycleBuffer();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/fc1244eb/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
index ced1a33..7169796 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.io.network.partition;
import org.apache.flink.configuration.AkkaOptions;
-import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.execution.Environment;
@@ -32,41 +31,51 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
-import org.apache.flink.runtime.testingUtils.TestingCluster;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.testutils.category.New;
import org.apache.flink.util.TestLogger;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
+@Category(New.class)
public class PartialConsumePipelinedResultTest extends TestLogger {
// Test configuration
- private final static int NUMBER_OF_TMS = 1;
- private final static int NUMBER_OF_SLOTS_PER_TM = 1;
- private final static int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
+ private static final int NUMBER_OF_TMS = 1;
+ private static final int NUMBER_OF_SLOTS_PER_TM = 1;
+ private static final int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
- private final static int NUMBER_OF_NETWORK_BUFFERS = 128;
+ private static final int NUMBER_OF_NETWORK_BUFFERS = 128;
- private static TestingCluster flink;
+ private static MiniCluster flink;
@BeforeClass
public static void setUp() throws Exception {
final Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUMBER_OF_TMS);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUMBER_OF_SLOTS_PER_TM);
config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
- flink = new TestingCluster(config, true);
+ final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder()
+ .setConfiguration(config)
+ .setNumTaskManagers(NUMBER_OF_TMS)
+ .setNumSlotsPerTaskManager(NUMBER_OF_SLOTS_PER_TM)
+ .build();
+
+ flink = new MiniCluster(miniClusterConfiguration);
flink.start();
}
@AfterClass
public static void tearDown() throws Exception {
- flink.stop();
+ if (flink != null) {
+ flink.close();
+ }
}
/**
@@ -102,7 +111,7 @@ public class PartialConsumePipelinedResultTest extends TestLogger {
sender.setSlotSharingGroup(slotSharingGroup);
receiver.setSlotSharingGroup(slotSharingGroup);
- flink.submitJobAndWait(jobGraph, false, TestingUtils.TESTING_DURATION());
+ flink.executeJobBlocking(jobGraph);
}
// ---------------------------------------------------------------------------------------------