You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/07/09 12:46:14 UTC
[flink] 03/05: [hotfix][test] Deduplicate
NettyShuffleEnvironmentTest code
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 282646ec7e84b4dfe17d95a18585510577366a82
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Jul 9 14:33:27 2019 +0200
[hotfix][test] Deduplicate NettyShuffleEnvironmentTest code
---
.../io/network/NettyShuffleEnvironmentBuilder.java | 4 +-
.../io/network/NettyShuffleEnvironmentTest.java | 67 +++-------------------
2 files changed, 10 insertions(+), 61 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
index 32b3744..517be70 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
@@ -30,9 +30,11 @@ import org.apache.flink.runtime.util.EnvironmentInformation;
*/
public class NettyShuffleEnvironmentBuilder {
+ public static final int DEFAULT_NUM_NETWORK_BUFFERS = 1024;
+
private static final String[] DEFAULT_TEMP_DIRS = new String[] {EnvironmentInformation.getTemporaryFileDirectory()};
- private int numNetworkBuffers = 1024;
+ private int numNetworkBuffers = DEFAULT_NUM_NETWORK_BUFFERS;
private int networkBufferSize = 32 * 1024;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
index 11951f2..ab847ad 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
@@ -89,62 +89,7 @@ public class NettyShuffleEnvironmentTest extends TestLogger {
*/
@Test
public void testRegisterTaskUsesBoundedBuffers() throws Exception {
- final NettyShuffleEnvironment network = new NettyShuffleEnvironmentBuilder()
- .setIsCreditBased(enableCreditBasedFlowControl)
- .build();
-
- // result partitions
- ResultPartition rp1 = createPartition(network, ResultPartitionType.PIPELINED, 2);
- ResultPartition rp2 = createPartition(network, fileChannelManager, ResultPartitionType.BLOCKING, 2);
- ResultPartition rp3 = createPartition(network, ResultPartitionType.PIPELINED_BOUNDED, 2);
- ResultPartition rp4 = createPartition(network, ResultPartitionType.PIPELINED_BOUNDED, 8);
- final ResultPartition[] resultPartitions = new ResultPartition[] {rp1, rp2, rp3, rp4};
-
- // input gates
- SingleInputGate ig1 = createSingleInputGate(network, ResultPartitionType.PIPELINED, 2);
- SingleInputGate ig2 = createSingleInputGate(network, ResultPartitionType.BLOCKING, 2);
- SingleInputGate ig3 = createSingleInputGate(network, ResultPartitionType.PIPELINED_BOUNDED, 2);
- SingleInputGate ig4 = createSingleInputGate(network, ResultPartitionType.PIPELINED_BOUNDED, 8);
- final SingleInputGate[] inputGates = new SingleInputGate[] {ig1, ig2, ig3, ig4};
-
- Task.setupPartitionsAndGates(resultPartitions, inputGates);
-
- // verify buffer pools for the result partitions
- assertEquals(rp1.getNumberOfSubpartitions(), rp1.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(rp2.getNumberOfSubpartitions(), rp2.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(rp3.getNumberOfSubpartitions(), rp3.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(rp4.getNumberOfSubpartitions(), rp4.getBufferPool().getNumberOfRequiredMemorySegments());
-
- assertEquals(Integer.MAX_VALUE, rp1.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(Integer.MAX_VALUE, rp2.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(2 * 2 + 8, rp3.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(8 * 2 + 8, rp4.getBufferPool().getMaxNumberOfMemorySegments());
-
- // verify buffer pools for the input gates (NOTE: credit-based uses minimum required buffers
- // for exclusive buffers not managed by the buffer pool)
- assertEquals(enableCreditBasedFlowControl ? 0 : 2, ig1.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(enableCreditBasedFlowControl ? 0 : 2, ig2.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(enableCreditBasedFlowControl ? 0 : 2, ig3.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(enableCreditBasedFlowControl ? 0 : 8, ig4.getBufferPool().getNumberOfRequiredMemorySegments());
-
- assertEquals(Integer.MAX_VALUE, ig1.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(Integer.MAX_VALUE, ig2.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(enableCreditBasedFlowControl ? 8 : 2 * 2 + 8, ig3.getBufferPool().getMaxNumberOfMemorySegments());
- assertEquals(enableCreditBasedFlowControl ? 8 : 8 * 2 + 8, ig4.getBufferPool().getMaxNumberOfMemorySegments());
-
- int invokations = enableCreditBasedFlowControl ? 1 : 0;
- verify(ig1, times(invokations)).assignExclusiveSegments();
- verify(ig2, times(invokations)).assignExclusiveSegments();
- verify(ig3, times(invokations)).assignExclusiveSegments();
- verify(ig4, times(invokations)).assignExclusiveSegments();
-
- for (ResultPartition rp : resultPartitions) {
- rp.release();
- }
- for (SingleInputGate ig : inputGates) {
- ig.close();
- }
- network.close();
+ testRegisterTaskWithLimitedBuffers(NettyShuffleEnvironmentBuilder.DEFAULT_NUM_NETWORK_BUFFERS, false);
}
/**
@@ -164,7 +109,7 @@ public class NettyShuffleEnvironmentTest extends TestLogger {
bufferCount = 10 + 10 * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue();
}
- testRegisterTaskWithLimitedBuffers(bufferCount);
+ testRegisterTaskWithLimitedBuffers(bufferCount, true);
}
/**
@@ -185,10 +130,10 @@ public class NettyShuffleEnvironmentTest extends TestLogger {
expectedException.expect(IOException.class);
expectedException.expectMessage("Insufficient number of network buffers");
- testRegisterTaskWithLimitedBuffers(bufferCount);
+ testRegisterTaskWithLimitedBuffers(bufferCount, true);
}
- private void testRegisterTaskWithLimitedBuffers(int bufferPoolSize) throws Exception {
+ private void testRegisterTaskWithLimitedBuffers(int bufferPoolSize, boolean assertNumBuffers) throws Exception {
final NettyShuffleEnvironment network = new NettyShuffleEnvironmentBuilder()
.setNumNetworkBuffers(bufferPoolSize)
.setIsCreditBased(enableCreditBasedFlowControl)
@@ -238,7 +183,9 @@ public class NettyShuffleEnvironmentTest extends TestLogger {
for (ResultPartition rp : resultPartitions) {
assertEquals(rp.getNumberOfSubpartitions(), rp.getBufferPool().getNumberOfRequiredMemorySegments());
- assertEquals(rp.getNumberOfSubpartitions(), rp.getBufferPool().getNumBuffers());
+ if (assertNumBuffers) {
+ assertEquals(rp.getNumberOfSubpartitions(), rp.getBufferPool().getNumBuffers());
+ }
}
// verify buffer pools for the input gates (NOTE: credit-based uses minimum required buffers