flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [6/7] flink git commit: [FLINK-4545] [network] Allow LocalBufferPool to limited the number of used buffers
Date Fri, 10 Mar 2017 13:42:48 GMT
[FLINK-4545] [network] Allow LocalBufferPool to limited the number of used buffers

Use "a * <number of channels> + b" buffers for bounded pipelined partitions:

Default: a = 2, b = 8
* 1 buffer for in-flight data in the subpartition/input channel
* 1 buffer for parallel serialization
* + some extra buffers (8 seems a good default given bandwidth-delay products of current networks)

This closes #3480


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/11e2aa6d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/11e2aa6d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/11e2aa6d

Branch: refs/heads/master
Commit: 11e2aa6dcdbf42992cda57a5b50d5c29b4facf2d
Parents: e5e6da0
Author: Nico Kruber <nico@data-artisans.com>
Authored: Fri Feb 10 14:53:09 2017 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Fri Mar 10 13:30:58 2017 +0100

----------------------------------------------------------------------
 .../flink/configuration/TaskManagerOptions.java |  14 ++
 .../BackPressureStatsTrackerITCase.java         |   2 +-
 .../InputGateDeploymentDescriptor.java          |  15 ++
 .../runtime/executiongraph/ExecutionVertex.java |   7 +-
 .../runtime/io/network/NetworkEnvironment.java  |  23 ++-
 .../runtime/io/network/buffer/BufferPool.java   |  10 +-
 .../io/network/buffer/BufferPoolFactory.java    |   7 +-
 .../io/network/buffer/LocalBufferPool.java      |  57 ++++++-
 .../io/network/buffer/NetworkBufferPool.java    |  64 +++++--
 .../netty/PartitionRequestServerHandler.java    |   2 +-
 .../io/network/partition/ResultPartition.java   |  10 ++
 .../network/partition/ResultPartitionType.java  |  32 +++-
 .../partition/consumer/SingleInputGate.java     |  18 +-
 .../taskexecutor/TaskManagerServices.java       |   4 +-
 .../TaskManagerServicesConfiguration.java       |  17 +-
 .../NetworkEnvironmentConfiguration.scala       |   2 +
 .../io/network/MockNetworkEnvironment.java      |   2 +-
 .../io/network/NetworkEnvironmentTest.java      | 165 ++++++++++++++++++
 .../io/network/api/writer/RecordWriterTest.java |   2 +-
 .../network/buffer/BufferPoolFactoryTest.java   | 170 ++++++++++++++++---
 .../io/network/buffer/LocalBufferPoolTest.java  |  58 +++++++
 .../network/buffer/NetworkBufferPoolTest.java   |  96 ++++++++---
 .../partition/InputGateConcurrentTest.java      |   6 +-
 .../partition/InputGateFairnessTest.java        |   3 +-
 .../consumer/LocalInputChannelTest.java         |   6 +-
 .../partition/consumer/SingleInputGateTest.java |  24 ++-
 .../partition/consumer/TestSingleInputGate.java |   2 +
 .../partition/consumer/UnionInputGateTest.java  |  15 +-
 ...askManagerComponentsStartupShutdownTest.java |   7 +-
 .../runtime/taskmanager/TaskManagerTest.java    |  29 +++-
 .../api/graph/StreamingJobGraphGenerator.java   |   6 +-
 .../graph/StreamingJobGraphGeneratorTest.java   |  21 ++-
 .../io/BarrierBufferMassiveRandomTest.java      |   4 +-
 33 files changed, 790 insertions(+), 110 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index b7ee20a..b891e35 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -53,6 +53,20 @@ public class TaskManagerOptions {
 			key("taskmanager.net.request-backoff.max")
 			.defaultValue(10000);
 
+	/**
+	 * Number of network buffers to use for each outgoing/ingoing channel (subpartition/input channel).
+	 *
+	 * Reasoning: 1 buffer for in-flight data in the subpartition + 1 buffer for parallel serialization
+	 */
+	public static final ConfigOption<Integer> NETWORK_BUFFERS_PER_CHANNEL =
+		key("taskmanager.net.memory.buffers-per-channel")
+			.defaultValue(2);
+
+	/** Number of extra network buffers to use for each outgoing/ingoing gate (result partition/input gate). */
+	public static final ConfigOption<Integer> NETWORK_EXTRA_BUFFERS_PER_GATE =
+		key("taskmanager.net.memory.extra-buffers-per-gate")
+			.defaultValue(8);
+
 	// ------------------------------------------------------------------------
 	//  Task Options
 	// ------------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
index 1943129..46f8be6 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
@@ -108,7 +108,7 @@ public class BackPressureStatsTrackerITCase extends TestLogger {
 			//
 			// 1) Consume all buffers at first (no buffers for the test task)
 			//
-			testBufferPool = networkBufferPool.createBufferPool(1);
+			testBufferPool = networkBufferPool.createBufferPool(1, Integer.MAX_VALUE);
 			final List<Buffer> buffers = new ArrayList<>();
 			while (true) {
 				Buffer buffer = testBufferPool.requestBuffer();

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java
index dde1ed7..9bf724a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/InputGateDeploymentDescriptor.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.deployment;
 
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
 import org.apache.flink.runtime.jobgraph.DistributionPattern;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
@@ -46,6 +47,9 @@ public class InputGateDeploymentDescriptor implements Serializable {
 	 */
 	private final IntermediateDataSetID consumedResultId;
 
+	/** The type of the partition the input gate is going to consume. */
+	private final ResultPartitionType consumedPartitionType;
+
 	/**
 	 * The index of the consumed subpartition of each consumed partition. This index depends on the
 	 * {@link DistributionPattern} and the subtask indices of the producing and consuming task.
@@ -57,10 +61,12 @@ public class InputGateDeploymentDescriptor implements Serializable {
 
 	public InputGateDeploymentDescriptor(
 			IntermediateDataSetID consumedResultId,
+			ResultPartitionType consumedPartitionType,
 			int consumedSubpartitionIndex,
 			InputChannelDeploymentDescriptor[] inputChannels) {
 
 		this.consumedResultId = checkNotNull(consumedResultId);
+		this.consumedPartitionType = checkNotNull(consumedPartitionType);
 
 		checkArgument(consumedSubpartitionIndex >= 0);
 		this.consumedSubpartitionIndex = consumedSubpartitionIndex;
@@ -72,6 +78,15 @@ public class InputGateDeploymentDescriptor implements Serializable {
 		return consumedResultId;
 	}
 
+	/**
+	 * Returns the type of this input channel's consumed result partition.
+	 *
+	 * @return consumed result partition type
+	 */
+	public ResultPartitionType getConsumedPartitionType() {
+		return consumedPartitionType;
+	}
+
 	public int getConsumedSubpartitionIndex() {
 		return consumedSubpartitionIndex;
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index ca8e07c..21af73a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.instance.SimpleSlot;
 import org.apache.flink.runtime.instance.SlotProvider;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.jobgraph.DistributionPattern;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
@@ -687,9 +688,11 @@ public class ExecutionVertex implements AccessExecutionVertex, Archiveable<Archi
 
 			int queueToRequest = subTaskIndex % numConsumerEdges;
 
-			IntermediateDataSetID resultId = edges[0].getSource().getIntermediateResult().getId();
+			IntermediateResult consumedIntermediateResult = edges[0].getSource().getIntermediateResult();
+			final IntermediateDataSetID resultId = consumedIntermediateResult.getId();
+			final ResultPartitionType partitionType = consumedIntermediateResult.getResultType();
 
-			consumedPartitions.add(new InputGateDeploymentDescriptor(resultId, queueToRequest, partitions));
+			consumedPartitions.add(new InputGateDeploymentDescriptor(resultId, partitionType, queueToRequest, partitions));
 		}
 
 		SerializedValue<JobInformation> serializedJobInformation = getExecutionGraph().getSerializedJobInformation();

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index 8e85ffe..4d4b305 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -73,6 +73,11 @@ public class NetworkEnvironment {
 
 	private final int partitionRequestMaxBackoff;
 
+	/** Number of network buffers to use for each outgoing/ingoing channel (subpartition/input channel). */
+	private final int networkBuffersPerChannel;
+	/** Number of extra network buffers to use for each outgoing/ingoing gate (result partition/input gate). */
+	private final int extraNetworkBuffersPerGate;
+
 	private boolean isShutdown;
 
 	public NetworkEnvironment(
@@ -84,7 +89,9 @@ public class NetworkEnvironment {
 			KvStateServer kvStateServer,
 			IOMode defaultIOMode,
 			int partitionRequestInitialBackoff,
-			int partitionRequestMaxBackoff) {
+			int partitionRequestMaxBackoff,
+			int networkBuffersPerChannel,
+			int extraNetworkBuffersPerGate) {
 
 		this.networkBufferPool = checkNotNull(networkBufferPool);
 		this.connectionManager = checkNotNull(connectionManager);
@@ -100,6 +107,8 @@ public class NetworkEnvironment {
 		this.partitionRequestMaxBackoff = partitionRequestMaxBackoff;
 
 		isShutdown = false;
+		this.networkBuffersPerChannel = networkBuffersPerChannel;
+		this.extraNetworkBuffersPerGate = extraNetworkBuffersPerGate;
 	}
 
 	// --------------------------------------------------------------------------------------------
@@ -171,7 +180,11 @@ public class NetworkEnvironment {
 				BufferPool bufferPool = null;
 
 				try {
-					bufferPool = networkBufferPool.createBufferPool(partition.getNumberOfSubpartitions());
+					int maxNumberOfMemorySegments = partition.getPartitionType().isBounded() ?
+						partition.getNumberOfSubpartitions() * networkBuffersPerChannel +
+							extraNetworkBuffersPerGate : Integer.MAX_VALUE;
+					bufferPool = networkBufferPool.createBufferPool(partition.getNumberOfSubpartitions(),
+							maxNumberOfMemorySegments);
 					partition.registerBufferPool(bufferPool);
 
 					resultPartitionManager.registerResultPartition(partition);
@@ -198,7 +211,11 @@ public class NetworkEnvironment {
 				BufferPool bufferPool = null;
 
 				try {
-					bufferPool = networkBufferPool.createBufferPool(gate.getNumberOfInputChannels());
+					int maxNumberOfMemorySegments = gate.getConsumedPartitionType().isBounded() ?
+						gate.getNumberOfInputChannels() * networkBuffersPerChannel +
+							extraNetworkBuffersPerGate : Integer.MAX_VALUE;
+					bufferPool = networkBufferPool.createBufferPool(gate.getNumberOfInputChannels(),
+						maxNumberOfMemorySegments);
 					gate.setBufferPool(bufferPool);
 				} catch (Throwable t) {
 					if (bufferPool != null) {

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java
index 8784b14..a4928ed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPool.java
@@ -50,6 +50,13 @@ public interface BufferPool extends BufferProvider, BufferRecycler {
 	int getNumberOfRequiredMemorySegments();
 
 	/**
+	 * Returns the maximum number of memory segments this buffer pool should use
+	 *
+	 * @return maximum number of memory segments to use or <tt>-1</tt> if unlimited
+	 */
+	int getMaxNumberOfMemorySegments();
+
+	/**
 	 * Returns the current size of this buffer pool.
 	 *
 	 * <p> The size of the buffer pool can change dynamically at runtime.
@@ -59,7 +66,7 @@ public interface BufferPool extends BufferProvider, BufferRecycler {
 	/**
 	 * Sets the current size of this buffer pool.
 	 *
-	 * <p> The size needs to be greater or equals to the guaranteed number of memory segments.
+	 * <p> The size needs to be greater or equal to the guaranteed number of memory segments.
 	 */
 	void setNumBuffers(int numBuffers) throws IOException;
 
@@ -72,5 +79,4 @@ public interface BufferPool extends BufferProvider, BufferRecycler {
 	 * Returns the number of used buffers of this buffer pool.
 	 */
 	int bestEffortGetNumOfUsedBuffers();
-
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java
index e953158..ffed432 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactory.java
@@ -30,8 +30,13 @@ public interface BufferPoolFactory {
 	 * buffers.
 	 *
 	 * <p> The buffer pool is of dynamic size with at least <tt>numRequiredBuffers</tt> buffers.
+	 *
+	 * @param numRequiredBuffers
+	 * 		minimum number of network buffers in this pool
+	 * @param maxUsedBuffers
+	 * 		maximum number of network buffers this pool offers
 	 */
-	BufferPool createBufferPool(int numRequiredBuffers) throws IOException;
+	BufferPool createBufferPool(int numRequiredBuffers, int maxUsedBuffers) throws IOException;
 
 	/**
 	 * Destroy callback for updating factory book keeping.

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
index d6a4cf7..a587997 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
@@ -20,6 +20,8 @@ package org.apache.flink.runtime.io.network.buffer;
 
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.util.event.EventListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.ArrayDeque;
@@ -44,6 +46,7 @@ import static org.apache.flink.util.Preconditions.checkState;
  * match its new size.
  */
 class LocalBufferPool implements BufferPool {
+	private static final Logger LOG = LoggerFactory.getLogger(LocalBufferPool.class);
 
 	/** Global network buffer pool to get buffers from. */
 	private final NetworkBufferPool networkBufferPool;
@@ -63,6 +66,9 @@ class LocalBufferPool implements BufferPool {
 	 */
 	private final Queue<EventListener<Buffer>> registeredListeners = new ArrayDeque<EventListener<Buffer>>();
 
+	/** Maximum number of network buffers to allocate. */
+	private final int maxNumberOfMemorySegments;
+
 	/** The current size of this pool */
 	private int currentPoolSize;
 
@@ -86,9 +92,37 @@ class LocalBufferPool implements BufferPool {
 	 * 		minimum number of network buffers
 	 */
 	LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments) {
+		this(networkBufferPool, numberOfRequiredMemorySegments, Integer.MAX_VALUE);
+	}
+
+	/**
+	 * Local buffer pool based on the given <tt>networkBufferPool</tt> with a minimal and maximal
+	 * number of network buffers being available.
+	 *
+	 * @param networkBufferPool
+	 * 		global network buffer pool to get buffers from
+	 * @param numberOfRequiredMemorySegments
+	 * 		minimum number of network buffers
+	 * @param maxNumberOfMemorySegments
+	 * 		maximum number of network buffers to allocate
+	 */
+	LocalBufferPool(NetworkBufferPool networkBufferPool, int numberOfRequiredMemorySegments,
+			int maxNumberOfMemorySegments) {
+		checkArgument(maxNumberOfMemorySegments >= numberOfRequiredMemorySegments,
+			"Maximum number of memory segments (%s) should not be smaller than minimum (%s).",
+			maxNumberOfMemorySegments, numberOfRequiredMemorySegments);
+
+		checkArgument(maxNumberOfMemorySegments > 0,
+			"Maximum number of memory segments (%s) should be larger than 0.",
+			maxNumberOfMemorySegments);
+
+		LOG.debug("Using a local buffer pool with {}-{} buffers",
+			numberOfRequiredMemorySegments, maxNumberOfMemorySegments);
+
 		this.networkBufferPool = networkBufferPool;
 		this.numberOfRequiredMemorySegments = numberOfRequiredMemorySegments;
 		this.currentPoolSize = numberOfRequiredMemorySegments;
+		this.maxNumberOfMemorySegments = maxNumberOfMemorySegments;
 	}
 
 	// ------------------------------------------------------------------------
@@ -113,6 +147,11 @@ class LocalBufferPool implements BufferPool {
 	}
 
 	@Override
+	public int getMaxNumberOfMemorySegments() {
+		return maxNumberOfMemorySegments;
+	}
+
+	@Override
 	public int getNumberOfAvailableMemorySegments() {
 		synchronized (availableMemorySegments) {
 			return availableMemorySegments.size();
@@ -160,6 +199,7 @@ class LocalBufferPool implements BufferPool {
 
 			boolean askToRecycle = owner != null;
 
+			// fill availableMemorySegments with at least one element, wait if required
 			while (availableMemorySegments.isEmpty()) {
 				if (isDestroyed) {
 					throw new IllegalStateException("Buffer pool is destroyed.");
@@ -257,9 +297,14 @@ class LocalBufferPool implements BufferPool {
 	@Override
 	public void setNumBuffers(int numBuffers) throws IOException {
 		synchronized (availableMemorySegments) {
-			checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least " + numberOfRequiredMemorySegments + " buffers, but tried to set to " + numBuffers + ".");
+			checkArgument(numBuffers >= numberOfRequiredMemorySegments, "Buffer pool needs at least " +
+				numberOfRequiredMemorySegments + " buffers, but tried to set to " + numBuffers + ".");
 
-			currentPoolSize = numBuffers;
+			if (numBuffers > maxNumberOfMemorySegments) {
+				currentPoolSize = maxNumberOfMemorySegments;
+			} else {
+				currentPoolSize = numBuffers;
+			}
 
 			returnExcessMemorySegments();
 
@@ -274,7 +319,10 @@ class LocalBufferPool implements BufferPool {
 	@Override
 	public String toString() {
 		synchronized (availableMemorySegments) {
-			return String.format("[size: %d, required: %d, requested: %d, available: %d, listeners: %d, destroyed: %s]", currentPoolSize, numberOfRequiredMemorySegments, numberOfRequestedMemorySegments, availableMemorySegments.size(), registeredListeners.size(), isDestroyed);
+			return String.format(
+				"[size: %d, required: %d, requested: %d, available: %d, max: %d, listeners: %d, destroyed: %s]",
+				currentPoolSize, numberOfRequiredMemorySegments, numberOfRequestedMemorySegments,
+				availableMemorySegments.size(), maxNumberOfMemorySegments, registeredListeners.size(), isDestroyed);
 		}
 	}
 
@@ -296,8 +344,7 @@ class LocalBufferPool implements BufferPool {
 				return;
 			}
 
-			networkBufferPool.recycle(segment);
-			numberOfRequestedMemorySegments--;
+			returnMemorySegment(segment);
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
index 5345fbb..7668759 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPool.java
@@ -180,7 +180,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
 	// ------------------------------------------------------------------------
 
 	@Override
-	public BufferPool createBufferPool(int numRequiredBuffers) throws IOException {
+	public BufferPool createBufferPool(int numRequiredBuffers, int maxUsedBuffers) throws IOException {
 		// It is necessary to use a separate lock from the one used for buffer
 		// requests to ensure deadlock freedom for failure cases.
 		synchronized (factoryLock) {
@@ -205,7 +205,8 @@ public class NetworkBufferPool implements BufferPoolFactory {
 
 			// We are good to go, create a new buffer pool and redistribute
 			// non-fixed size buffers.
-			LocalBufferPool localBufferPool = new LocalBufferPool(this, numRequiredBuffers);
+			LocalBufferPool localBufferPool =
+				new LocalBufferPool(this, numRequiredBuffers, maxUsedBuffers);
 
 			allBufferPools.add(localBufferPool);
 
@@ -236,7 +237,7 @@ public class NetworkBufferPool implements BufferPoolFactory {
 
 	/**
 	 * Destroys all buffer pools that allocate their buffers from this
-	 * buffer pool (created via {@link #createBufferPool(int)}).
+	 * buffer pool (created via {@link #createBufferPool(int, int)}).
 	 */
 	public void destroyAllBufferPools() {
 		synchronized (factoryLock) {
@@ -258,27 +259,60 @@ public class NetworkBufferPool implements BufferPoolFactory {
 	private void redistributeBuffers() throws IOException {
 		assert Thread.holdsLock(factoryLock);
 
-		int numManagedBufferPools = allBufferPools.size();
+		// All buffers, which are not among the required ones
+		int numAvailableMemorySegment = totalNumberOfMemorySegments - numTotalRequiredBuffers;
 
-		if (numManagedBufferPools == 0) {
-			return; // necessary to avoid div by zero when no managed pools
+		if (numAvailableMemorySegment == 0) {
+			// in this case, we need to redistribute buffers so that every pool gets its minimum
+			for (LocalBufferPool bufferPool : allBufferPools) {
+				bufferPool.setNumBuffers(bufferPool.getNumberOfRequiredMemorySegments());
+			}
+			return;
 		}
 
-		// All buffers, which are not among the required ones
-		int numAvailableMemorySegment = totalNumberOfMemorySegments - numTotalRequiredBuffers;
+		/**
+		 * With buffer pools being potentially limited, let's distribute the available memory
+		 * segments based on the capacity of each buffer pool, i.e. the maximum number of segments
+		 * an unlimited buffer pool can take is numAvailableMemorySegment, for limited buffer pools
+		 * it may be less. Based on this and the sum of all these values (totalCapacity), we build
+		 * a ratio that we use to distribute the buffers.
+		 */
 
-		// Available excess (not required) buffers per pool
-		int numExcessBuffersPerPool = numAvailableMemorySegment / numManagedBufferPools;
+		int totalCapacity = 0;
+		for (LocalBufferPool bufferPool : allBufferPools) {
+			int excessMax = bufferPool.getMaxNumberOfMemorySegments() -
+				bufferPool.getNumberOfRequiredMemorySegments();
+			totalCapacity += Math.min(numAvailableMemorySegment, excessMax);
+		}
 
-		// Distribute leftover buffers in round robin fashion
-		int numLeftoverBuffers = numAvailableMemorySegment % numManagedBufferPools;
+		// no capacity to receive additional buffers?
+		if (totalCapacity == 0) {
+			return; // necessary to avoid div by zero when nothing to re-distribute
+		}
 
-		int bufferPoolIndex = 0;
+		int memorySegmentsToDistribute = Math.min(numAvailableMemorySegment, totalCapacity);
 
+		int totalPartsUsed = 0; // of totalCapacity
+		int numDistributedMemorySegment = 0;
 		for (LocalBufferPool bufferPool : allBufferPools) {
-			int leftoverBuffers = bufferPoolIndex++ < numLeftoverBuffers ? 1 : 0;
+			int excessMax = bufferPool.getMaxNumberOfMemorySegments() -
+				bufferPool.getNumberOfRequiredMemorySegments();
+
+			// shortcut
+			if (excessMax == 0) {
+				continue;
+			}
+
+			totalPartsUsed += Math.min(numAvailableMemorySegment, excessMax);
 
-			bufferPool.setNumBuffers(bufferPool.getNumberOfRequiredMemorySegments() + numExcessBuffersPerPool + leftoverBuffers);
+			// avoid remaining buffers by looking at the total capacity that should have been
+			// re-distributed up until here
+			int mySize = memorySegmentsToDistribute * totalPartsUsed / totalCapacity - numDistributedMemorySegment;
+			numDistributedMemorySegment += mySize;
+			bufferPool.setNumBuffers(bufferPool.getNumberOfRequiredMemorySegments() + mySize);
 		}
+
+		assert (totalPartsUsed == totalCapacity);
+		assert (numDistributedMemorySegment == memorySegmentsToDistribute);
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
index 36c1234..6f56877 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestServerHandler.java
@@ -67,7 +67,7 @@ class PartitionRequestServerHandler extends SimpleChannelInboundHandler<NettyMes
 	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
 		super.channelRegistered(ctx);
 
-		bufferPool = networkBufferPool.createBufferPool(1);
+		bufferPool = networkBufferPool.createBufferPool(1, Integer.MAX_VALUE);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
index 3d92584..eb1418b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java
@@ -160,6 +160,7 @@ public class ResultPartition implements BufferPoolOwner {
 				break;
 
 			case PIPELINED:
+			case PIPELINED_BOUNDED:
 				for (int i = 0; i < subpartitions.length; i++) {
 					subpartitions[i] = new PipelinedSubpartition(i, this);
 				}
@@ -237,6 +238,15 @@ public class ResultPartition implements BufferPoolOwner {
 		return totalBuffers;
 	}
 
+	/**
+	 * Returns the type of this result partition.
+	 *
+	 * @return result partition type
+	 */
+	public ResultPartitionType getPartitionType() {
+		return partitionType;
+	}
+
 	// ------------------------------------------------------------------------
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
index 43d3a52..256387c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionType.java
@@ -20,9 +20,22 @@ package org.apache.flink.runtime.io.network.partition;
 
 public enum ResultPartitionType {
 
-	BLOCKING(false, false),
+	BLOCKING(false, false, false),
 
-	PIPELINED(true, true);
+	PIPELINED(true, true, false),
+
+	/**
+	 * Pipelined partitions with a bounded (local) buffer pool.
+	 *
+	 * For streaming jobs, a fixed limit on the buffer pool size should help avoid that too much
+	 * data is being buffered and checkpoint barriers are delayed. In contrast to limiting the
+	 * overall network buffer pool size, this, however, still allows to be flexible with regards
+	 * to the total number of partitions by selecting an appropriately big network buffer pool size.
+	 *
+	 * For batch jobs, it will be best to keep this unlimited ({@link #PIPELINED}) since there are
+	 * no checkpoint barriers.
+	 */
+	PIPELINED_BOUNDED(true, true, true);
 
 	/** Can the partition be consumed while being produced? */
 	private final boolean isPipelined;
@@ -30,12 +43,16 @@ public enum ResultPartitionType {
 	/** Does the partition produce back pressure when not consumed? */
 	private final boolean hasBackPressure;
 
+	/** Does this partition use a limited number of (network) buffers? */
+	private final boolean isBounded;
+
 	/**
 	 * Specifies the behaviour of an intermediate result partition at runtime.
 	 */
-	ResultPartitionType(boolean isPipelined, boolean hasBackPressure) {
+	ResultPartitionType(boolean isPipelined, boolean hasBackPressure, boolean isBounded) {
 		this.isPipelined = isPipelined;
 		this.hasBackPressure = hasBackPressure;
+		this.isBounded = isBounded;
 	}
 
 	public boolean hasBackPressure() {
@@ -49,4 +66,13 @@ public enum ResultPartitionType {
 	public boolean isPipelined() {
 		return isPipelined;
 	}
+
+	/**
+	 * Whether this partition uses a limited number of (network) buffers or not.
+	 *
+	 * @return <tt>true</tt> if the number of buffers should be bound to some limit
+	 */
+	public boolean isBounded() {
+		return isBounded;
+	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index d546559..afe8722 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
 import org.apache.flink.runtime.io.network.buffer.BufferPool;
 import org.apache.flink.runtime.io.network.buffer.BufferProvider;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.io.network.partition.consumer.InputChannel.BufferAndAvailability;
 import org.apache.flink.runtime.jobgraph.DistributionPattern;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
@@ -115,6 +116,9 @@ public class SingleInputGate implements InputGate {
 	 */
 	private final IntermediateDataSetID consumedResultId;
 
+	/** The type of the partition the input gate is consuming. */
+	private final ResultPartitionType consumedPartitionType;
+
 	/**
 	 * The index of the consumed subpartition of each consumed partition. This index depends on the
 	 * {@link DistributionPattern} and the subtask indices of the producing and consuming task.
@@ -166,6 +170,7 @@ public class SingleInputGate implements InputGate {
 		String owningTaskName,
 		JobID jobId,
 		IntermediateDataSetID consumedResultId,
+		final ResultPartitionType consumedPartitionType,
 		int consumedSubpartitionIndex,
 		int numberOfInputChannels,
 		TaskActions taskActions,
@@ -175,6 +180,7 @@ public class SingleInputGate implements InputGate {
 		this.jobId = checkNotNull(jobId);
 
 		this.consumedResultId = checkNotNull(consumedResultId);
+		this.consumedPartitionType = checkNotNull(consumedPartitionType);
 
 		checkArgument(consumedSubpartitionIndex >= 0);
 		this.consumedSubpartitionIndex = consumedSubpartitionIndex;
@@ -201,6 +207,15 @@ public class SingleInputGate implements InputGate {
 		return consumedResultId;
 	}
 
+	/**
+	 * Returns the type of this input channel's consumed result partition.
+	 *
+	 * @return consumed result partition type
+	 */
+	public ResultPartitionType getConsumedPartitionType() {
+		return consumedPartitionType;
+	}
+
 	BufferProvider getBufferProvider() {
 		return bufferPool;
 	}
@@ -571,6 +586,7 @@ public class SingleInputGate implements InputGate {
 		TaskIOMetricGroup metrics) {
 
 		final IntermediateDataSetID consumedResultId = checkNotNull(igdd.getConsumedResultId());
+		final ResultPartitionType consumedPartitionType = checkNotNull(igdd.getConsumedPartitionType());
 
 		final int consumedSubpartitionIndex = igdd.getConsumedSubpartitionIndex();
 		checkArgument(consumedSubpartitionIndex >= 0);
@@ -578,7 +594,7 @@ public class SingleInputGate implements InputGate {
 		final InputChannelDeploymentDescriptor[] icdd = checkNotNull(igdd.getInputChannelDeploymentDescriptors());
 
 		final SingleInputGate inputGate = new SingleInputGate(
-			owningTaskName, jobId, consumedResultId, consumedSubpartitionIndex,
+			owningTaskName, jobId, consumedResultId, consumedPartitionType, consumedSubpartitionIndex,
 			icdd.length, taskActions, metrics);
 
 		// Create the input channels. There is one input channel for each consumed partition.

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index 19c5c01..e3c8345 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -370,7 +370,9 @@ public class TaskManagerServices {
 			kvStateServer,
 			networkEnvironmentConfiguration.ioMode(),
 			networkEnvironmentConfiguration.partitionRequestInitialBackoff(),
-			networkEnvironmentConfiguration.partitionRequestMaxBackoff());
+			networkEnvironmentConfiguration.partitionRequestMaxBackoff(),
+			networkEnvironmentConfiguration.networkBuffersPerChannel(),
+			networkEnvironmentConfiguration.extraNetworkBuffersPerGate());
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
index 2c76372..8ad318a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
@@ -22,6 +22,7 @@ import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.IllegalConfigurationException;
 import org.apache.flink.configuration.QueryableStateOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.memory.HeapMemorySegment;
 import org.apache.flink.core.memory.HybridMemorySegment;
 import org.apache.flink.core.memory.MemorySegmentFactory;
@@ -311,13 +312,25 @@ public class TaskManagerServicesConfiguration {
 			ioMode = IOManager.IOMode.SYNC;
 		}
 
+		int initialRequestBackoff = configuration.getInteger(
+			TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL);
+		int maxRequestBackoff = configuration.getInteger(
+			TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX);
+
+		int buffersPerChannel = configuration.getInteger(
+			TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL);
+		int extraBuffersPerGate = configuration.getInteger(
+			TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
+
 		return new NetworkEnvironmentConfiguration(
 			numNetworkBuffers,
 			pageSize,
 			memType,
 			ioMode,
-			500,
-			3000,
+			initialRequestBackoff,
+			maxRequestBackoff,
+			buffersPerChannel,
+			extraBuffersPerGate,
 			nettyConfig);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
index 6c7ca3e..4ecfe59 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.scala
@@ -29,4 +29,6 @@ case class NetworkEnvironmentConfiguration(
     ioMode: IOMode,
     partitionRequestInitialBackoff : Int,
     partitionRequestMaxBackoff : Int,
+    networkBuffersPerChannel: Int,
+    extraNetworkBuffersPerGate: Int,
     nettyConfig: NettyConfig = null)

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/MockNetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/MockNetworkEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/MockNetworkEnvironment.java
index dcdf44c..3bbe6d5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/MockNetworkEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/MockNetworkEnvironment.java
@@ -36,7 +36,7 @@ public class MockNetworkEnvironment {
 
 	static {
 		try {
-			when(networkBufferPool.createBufferPool(anyInt())).thenReturn(mock(BufferPool.class));
+			when(networkBufferPool.createBufferPool(anyInt(), anyInt())).thenReturn(mock(BufferPool.class));
 			when(networkEnvironment.getNetworkBufferPool()).thenReturn(networkBufferPool);
 
 			when(networkEnvironment.getTaskEventDispatcher()).thenReturn(taskEventDispatcher);

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
new file mode 100644
index 0000000..b956691
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.memory.MemoryType;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+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.ResultPartition;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.query.KvStateRegistry;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.taskmanager.TaskActions;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Various tests for the {@link NetworkEnvironment} class.
+ */
+public class NetworkEnvironmentTest {
+	private final static int numBuffers = 1024;
+
+	private final static int memorySegmentSize = 128;
+
+	/**
+	 * Verifies that {@link NetworkEnvironment#registerTask(Task)} sets up (un)bounded buffer pool
+	 * instances for various types of input and output channels.
+	 */
+	@Test
+	public void testRegisterTaskUsesBoundedBuffers() throws Exception {
+
+		final NetworkEnvironment network = new NetworkEnvironment(
+			new NetworkBufferPool(numBuffers, memorySegmentSize, MemoryType.HEAP),
+			new LocalConnectionManager(),
+			new ResultPartitionManager(),
+			new TaskEventDispatcher(),
+			new KvStateRegistry(),
+			null,
+			IOManager.IOMode.SYNC,
+			0,
+			0,
+			2,
+			8);
+
+		// result partitions
+		ResultPartition rp1 = createResultPartition(ResultPartitionType.PIPELINED, 2);
+		ResultPartition rp2 = createResultPartition(ResultPartitionType.BLOCKING, 2);
+		ResultPartition rp3 = createResultPartition(ResultPartitionType.PIPELINED_BOUNDED, 2);
+		ResultPartition rp4 = createResultPartition(ResultPartitionType.PIPELINED_BOUNDED, 8);
+		final ResultPartition[] resultPartitions = new ResultPartition[] {rp1, rp2, rp3, rp4};
+		final ResultPartitionWriter[] resultPartitionWriters = new ResultPartitionWriter[] {
+			new ResultPartitionWriter(rp1), new ResultPartitionWriter(rp2),
+			new ResultPartitionWriter(rp3), new ResultPartitionWriter(rp4)};
+
+		// input gates
+		final SingleInputGate[] inputGates = new SingleInputGate[] {
+			createSingleInputGateMock(ResultPartitionType.PIPELINED, 2),
+			createSingleInputGateMock(ResultPartitionType.BLOCKING, 2),
+			createSingleInputGateMock(ResultPartitionType.PIPELINED_BOUNDED, 2),
+			createSingleInputGateMock(ResultPartitionType.PIPELINED_BOUNDED, 8)};
+
+		// overall task to register
+		Task task = mock(Task.class);
+		when(task.getProducedPartitions()).thenReturn(resultPartitions);
+		when(task.getAllWriters()).thenReturn(resultPartitionWriters);
+		when(task.getAllInputGates()).thenReturn(inputGates);
+
+		network.registerTask(task);
+
+		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());
+
+		network.shutdown();
+	}
+
+	/**
+	 * Helper to create simple {@link ResultPartition} instance for use by a {@link Task} inside
+	 * {@link NetworkEnvironment#registerTask(Task)}.
+	 *
+	 * @param partitionType
+	 * 		the produced partition type
+	 * @param channels
+	 * 		the nummer of output channels
+	 *
+	 * @return instance with minimal data set and some mocks so that it is useful for {@link
+	 * NetworkEnvironment#registerTask(Task)}
+	 */
+	private static ResultPartition createResultPartition(
+			final ResultPartitionType partitionType, final int channels) {
+		return new ResultPartition(
+			"TestTask-" + partitionType + ":" + channels,
+			mock(TaskActions.class),
+			new JobID(),
+			new ResultPartitionID(),
+			partitionType,
+			channels,
+			channels,
+			mock(ResultPartitionManager.class),
+			mock(ResultPartitionConsumableNotifier.class),
+			mock(IOManager.class),
+			false);
+	}
+
+	/**
+	 * Helper to create a mock of a {@link SingleInputGate} for use by a {@link Task} inside
+	 * {@link NetworkEnvironment#registerTask(Task)}.
+	 *
+	 * @param partitionType
+	 * 		the consumed partition type
+	 * @param channels
+	 * 		the nummer of input channels
+	 *
+	 * @return mock with minimal functionality necessary by {@link NetworkEnvironment#registerTask(Task)}
+	 */
+	private static SingleInputGate createSingleInputGateMock(
+			final ResultPartitionType partitionType, final int channels) {
+		SingleInputGate ig = mock(SingleInputGate.class);
+		when(ig.getConsumedPartitionType()).thenReturn(partitionType);
+		when(ig.getNumberOfInputChannels()).thenReturn(channels);
+		doAnswer(new Answer<Void>() {
+			@Override
+			public Void answer(final InvocationOnMock invocation) throws Throwable {
+				BufferPool bp = invocation.getArgumentAt(0, BufferPool.class);
+				if (partitionType == ResultPartitionType.PIPELINED_BOUNDED) {
+					assertEquals(channels * 2 + 8, bp.getMaxNumberOfMemorySegments());
+				} else {
+					assertEquals(Integer.MAX_VALUE, bp.getMaxNumberOfMemorySegments());
+				}
+				return null;
+			}
+		}).when(ig).setBufferPool(any(BufferPool.class));
+		return ig;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java
----------------------------------------------------------------------
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 ca1d0a5..7d83fb5 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
@@ -177,7 +177,7 @@ public class RecordWriterTest {
 
 		try {
 			buffers = new NetworkBufferPool(1, 1024, MemoryType.HEAP);
-			bufferPool = spy(buffers.createBufferPool(1));
+			bufferPool = spy(buffers.createBufferPool(1, Integer.MAX_VALUE));
 
 			ResultPartitionWriter partitionWriter = mock(ResultPartitionWriter.class);
 			when(partitionWriter.getBufferProvider()).thenReturn(checkNotNull(bufferPool));

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactoryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactoryTest.java
index 49808c9..ce76a6d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferPoolFactoryTest.java
@@ -29,6 +29,7 @@ import java.util.List;
 import java.util.Random;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.fail;
 
 public class BufferPoolFactoryTest {
@@ -48,61 +49,190 @@ public class BufferPoolFactoryTest {
 	public void verifyAllBuffersReturned() {
 		String msg = "Did not return all buffers to network buffer pool after test.";
 		assertEquals(msg, numBuffers, networkBufferPool.getNumberOfAvailableMemorySegments());
+		// in case buffers have actually been requested, we must release them again
+		networkBufferPool.destroy();
 	}
 
 	@Test(expected = IOException.class)
 	public void testRequireMoreThanPossible() throws IOException {
-		networkBufferPool.createBufferPool(networkBufferPool.getTotalNumberOfMemorySegments() * 2);
+		networkBufferPool.createBufferPool(networkBufferPool.getTotalNumberOfMemorySegments() * 2, Integer.MAX_VALUE);
+	}
+
+	@Test
+	public void testBoundedPools() throws IOException {
+		BufferPool lbp = networkBufferPool.createBufferPool(1, 1);
+		assertEquals(1, lbp.getNumBuffers());
+
+		lbp = networkBufferPool.createBufferPool(1, 2);
+		assertEquals(2, lbp.getNumBuffers());
 	}
 
 	@Test
 	public void testSingleManagedPoolGetsAll() throws IOException {
-		BufferPool lbp = networkBufferPool.createBufferPool(1);
+		BufferPool lbp = networkBufferPool.createBufferPool(1, Integer.MAX_VALUE);
 
 		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments(), lbp.getNumBuffers());
 	}
 
 	@Test
+	public void testSingleManagedPoolGetsAllExceptFixedOnes() throws IOException {
+		BufferPool fixed = networkBufferPool.createBufferPool(24, 24);
+
+		BufferPool lbp = networkBufferPool.createBufferPool(1, Integer.MAX_VALUE);
+
+		assertEquals(24, fixed.getNumBuffers());
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() - fixed.getNumBuffers(), lbp.getNumBuffers());
+	}
+
+	@Test
 	public void testUniformDistribution() throws IOException {
-		BufferPool first = networkBufferPool.createBufferPool(0);
-		BufferPool second = networkBufferPool.createBufferPool(0);
+		BufferPool first = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments(), first.getNumBuffers());
 
+		BufferPool second = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
 		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, first.getNumBuffers());
 		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, second.getNumBuffers());
 	}
 
+	/**
+	 * Tests that buffers, once given to an initial buffer pool, get re-distributed to a second one
+	 * in case both buffer pools request half of the available buffer count.
+	 */
 	@Test
-	public void testAllDistributed() {
-		Random random = new Random();
+	public void testUniformDistributionAllBuffers() throws IOException {
+		BufferPool first = networkBufferPool
+			.createBufferPool(networkBufferPool.getTotalNumberOfMemorySegments() / 2, Integer.MAX_VALUE);
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments(), first.getNumBuffers());
 
-		try {
-			List<BufferPool> pools = new ArrayList<BufferPool>();
+		BufferPool second = networkBufferPool
+			.createBufferPool(networkBufferPool.getTotalNumberOfMemorySegments() / 2, Integer.MAX_VALUE);
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, first.getNumBuffers());
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2,
+			second.getNumBuffers());
+	}
 
-			int numPools = numBuffers / 32;
-			for (int i = 0; i < numPools; i++) {
-				pools.add(networkBufferPool.createBufferPool(random.nextInt(7 + 1)));
+	@Test
+	public void testUniformDistributionBounded1() throws IOException {
+		BufferPool first = networkBufferPool.createBufferPool(0, networkBufferPool.getTotalNumberOfMemorySegments());
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments(), first.getNumBuffers());
+
+		BufferPool second = networkBufferPool.createBufferPool(0, networkBufferPool.getTotalNumberOfMemorySegments());
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, first.getNumBuffers());
+		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, second.getNumBuffers());
+	}
+
+	@Test
+	public void testUniformDistributionBounded2() throws IOException {
+		BufferPool first = networkBufferPool.createBufferPool(0, 10);
+		assertEquals(10, first.getNumBuffers());
+
+		BufferPool second = networkBufferPool.createBufferPool(0, 10);
+		assertEquals(10, first.getNumBuffers());
+		assertEquals(10, second.getNumBuffers());
+	}
+
+	@Test
+	public void testUniformDistributionBounded3() throws IOException {
+		NetworkBufferPool globalPool = new NetworkBufferPool(3, 128, MemoryType.HEAP);
+		BufferPool first = globalPool.createBufferPool(0, 10);
+		assertEquals(3, first.getNumBuffers());
+
+		BufferPool second = globalPool.createBufferPool(0, 10);
+		// the order of which buffer pool received 2 or 1 buffer is undefined
+		assertEquals(3, first.getNumBuffers() + second.getNumBuffers());
+		assertNotEquals(3, first.getNumBuffers());
+		assertNotEquals(3, second.getNumBuffers());
+
+		BufferPool third = globalPool.createBufferPool(0, 10);
+		assertEquals(1, first.getNumBuffers());
+		assertEquals(1, second.getNumBuffers());
+		assertEquals(1, third.getNumBuffers());
+
+		// similar to #verifyAllBuffersReturned()
+		String msg = "Did not return all buffers to network buffer pool after test.";
+		assertEquals(msg, 3, globalPool.getNumberOfAvailableMemorySegments());
+		// in case buffers have actually been requested, we must release them again
+		globalPool.destroy();
+	}
+
+	@Test
+	public void testBufferRedistributionMixed1() throws IOException {
+		// try multiple times for various orders during redistribution
+		for (int i = 0; i < 1_000; ++i) {
+			BufferPool first = networkBufferPool.createBufferPool(0, 10);
+			assertEquals(10, first.getNumBuffers());
+
+			BufferPool second = networkBufferPool.createBufferPool(0, 10);
+			assertEquals(10, first.getNumBuffers());
+			assertEquals(10, second.getNumBuffers());
+
+			BufferPool third = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
+			// note: exact buffer distribution depends on the order during the redistribution
+			for (BufferPool bp : new BufferPool[] {first, second, third}) {
+				int size = networkBufferPool.getTotalNumberOfMemorySegments() *
+					Math.min(networkBufferPool.getTotalNumberOfMemorySegments(),
+						bp.getMaxNumberOfMemorySegments()) /
+					(networkBufferPool.getTotalNumberOfMemorySegments() + 20);
+				if (bp.getNumBuffers() != size && bp.getNumBuffers() != (size + 1)) {
+					fail("wrong buffer pool size after redistribution: " + bp.getNumBuffers());
+				}
 			}
 
-			int numDistributedBuffers = 0;
-			for (BufferPool pool : pools) {
-				numDistributedBuffers += pool.getNumBuffers();
+			BufferPool fourth = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
+			// note: exact buffer distribution depends on the order during the redistribution
+			for (BufferPool bp : new BufferPool[] {first, second, third, fourth}) {
+				int size = networkBufferPool.getTotalNumberOfMemorySegments() *
+					Math.min(networkBufferPool.getTotalNumberOfMemorySegments(),
+						bp.getMaxNumberOfMemorySegments()) /
+					(2 * networkBufferPool.getTotalNumberOfMemorySegments() + 20);
+				if (bp.getNumBuffers() != size && bp.getNumBuffers() != (size + 1)) {
+					fail("wrong buffer pool size after redistribution: " + bp.getNumBuffers());
+				}
 			}
 
-			assertEquals(numBuffers, numDistributedBuffers);
+			verifyAllBuffersReturned();
+			setupNetworkBufferPool();
 		}
-		catch (Throwable t) {
-			t.printStackTrace();
-			fail(t.getMessage());
+	}
+
+	@Test
+	public void testAllDistributed() throws IOException {
+		// try multiple times for various orders during redistribution
+		for (int i = 0; i < 1_000; ++i) {
+			Random random = new Random();
+
+			List<BufferPool> pools = new ArrayList<BufferPool>();
+
+			int numPools = numBuffers / 32;
+			long maxTotalUsed = 0;
+			for (int j = 0; j < numPools; j++) {
+				int numRequiredBuffers = random.nextInt(7 + 1);
+				// make unbounded buffers more likely:
+				int maxUsedBuffers = random.nextBoolean() ? Integer.MAX_VALUE :
+					Math.max(1, random.nextInt(10) + numRequiredBuffers);
+				pools.add(networkBufferPool.createBufferPool(numRequiredBuffers, maxUsedBuffers));
+				maxTotalUsed = Math.min(numBuffers, maxTotalUsed + maxUsedBuffers);
+
+				// after every iteration, all buffers (up to maxTotalUsed) must be distributed
+				int numDistributedBuffers = 0;
+				for (BufferPool pool : pools) {
+					numDistributedBuffers += pool.getNumBuffers();
+				}
+				assertEquals(maxTotalUsed, numDistributedBuffers);
+			}
+
+			verifyAllBuffersReturned();
+			setupNetworkBufferPool();
 		}
 	}
 
 	@Test
 	public void testCreateDestroy() throws IOException {
-		BufferPool first = networkBufferPool.createBufferPool(0);
+		BufferPool first = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
 
 		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments(), first.getNumBuffers());
 
-		BufferPool second = networkBufferPool.createBufferPool(0);
+		BufferPool second = networkBufferPool.createBufferPool(0, Integer.MAX_VALUE);
 
 		assertEquals(networkBufferPool.getTotalNumberOfMemorySegments() / 2, first.getNumBuffers());
 

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index 93731e1..a186d56 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -328,6 +328,64 @@ public class LocalBufferPoolTest {
 		}
 	}
 
+	@Test
+	public void testBoundedBuffer() throws Exception {
+		localBufferPool.lazyDestroy();
+
+		localBufferPool = new LocalBufferPool(networkBufferPool, 1, 2);
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertEquals(2, localBufferPool.getMaxNumberOfMemorySegments());
+
+		Buffer buffer1, buffer2;
+
+		// check min number of buffers:
+		localBufferPool.setNumBuffers(1);
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer1 = localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNull(localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		buffer1.recycle();
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+
+		// check max number of buffers:
+		localBufferPool.setNumBuffers(2);
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer1 = localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer2 = localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNull(localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		buffer1.recycle();
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+		buffer2.recycle();
+		assertEquals(2, localBufferPool.getNumberOfAvailableMemorySegments());
+
+		// try to set too large buffer size:
+		localBufferPool.setNumBuffers(3);
+		assertEquals(2, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer1 = localBufferPool.requestBuffer());
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer2 = localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNull(localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		buffer1.recycle();
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+		buffer2.recycle();
+		assertEquals(2, localBufferPool.getNumberOfAvailableMemorySegments());
+
+		// decrease size again
+		localBufferPool.setNumBuffers(1);
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNotNull(buffer1 = localBufferPool.requestBuffer());
+		assertEquals(0, localBufferPool.getNumberOfAvailableMemorySegments());
+		assertNull(localBufferPool.requestBuffer());
+		buffer1.recycle();
+		assertEquals(1, localBufferPool.getNumberOfAvailableMemorySegments());
+	}
+
 	// ------------------------------------------------------------------------
 	// Helpers
 	// ------------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
index ab32685..7c6a543 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
@@ -21,6 +21,8 @@ package org.apache.flink.runtime.io.network.buffer;
 import org.apache.flink.core.memory.MemoryType;
 import org.junit.Test;
 
+import java.util.ArrayList;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -47,7 +49,23 @@ public class NetworkBufferPoolTest {
 			assertTrue(globalPool.isDestroyed());
 
 			try {
-				globalPool.createBufferPool(2);
+				globalPool.createBufferPool(2, 2);
+				fail("Should throw an IllegalStateException");
+			}
+			catch (IllegalStateException e) {
+				// yippie!
+			}
+
+			try {
+				globalPool.createBufferPool(2, 10);
+				fail("Should throw an IllegalStateException");
+			}
+			catch (IllegalStateException e) {
+				// yippie!
+			}
+
+			try {
+				globalPool.createBufferPool(2, Integer.MAX_VALUE);
 				fail("Should throw an IllegalStateException");
 			}
 			catch (IllegalStateException e) {
@@ -60,39 +78,51 @@ public class NetworkBufferPoolTest {
 		}
 
 	}
+
 	@Test
 	public void testDestroyAll() {
 		try {
-			NetworkBufferPool globalPool = new NetworkBufferPool(8, 128, MemoryType.HEAP);
-
-			BufferPool lbp = globalPool.createBufferPool(5);
-
-			assertEquals(5, lbp.getNumberOfRequiredMemorySegments());
-
-			Buffer[] buffers = {
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer(),
-					lbp.requestBuffer()
-			};
-
-			for (Buffer b : buffers) {
-				assertNotNull(b);
-				assertNotNull(b.getMemorySegment());
+			NetworkBufferPool globalPool = new NetworkBufferPool(10, 128, MemoryType.HEAP);
+
+			BufferPool fixedPool = globalPool.createBufferPool(2, 2);
+			BufferPool boundedPool = globalPool.createBufferPool(0, 1);
+			BufferPool nonFixedPool = globalPool.createBufferPool(5, Integer.MAX_VALUE);
+
+			assertEquals(2, fixedPool.getNumberOfRequiredMemorySegments());
+			assertEquals(0, boundedPool.getNumberOfRequiredMemorySegments());
+			assertEquals(5, nonFixedPool.getNumberOfRequiredMemorySegments());
+
+			// actually, the buffer pool sizes may be different due to rounding and based on the internal order of
+			// the buffer pools - the total number of retrievable buffers should be equal to the number of buffers
+			// in the NetworkBufferPool though
+
+			ArrayList<Buffer> buffers = new ArrayList<>(globalPool.getTotalNumberOfMemorySegments());
+			collectBuffers:
+			for (int i = 0; i < 10; ++i) {
+				for (BufferPool bp : new BufferPool[] { fixedPool, boundedPool, nonFixedPool }) {
+					Buffer buffer = bp.requestBuffer();
+					if (buffer != null) {
+						assertNotNull(buffer.getMemorySegment());
+						buffers.add(buffer);
+						continue collectBuffers;
+					}
+				}
 			}
 
-			assertNull(lbp.requestBuffer());
+			assertEquals(globalPool.getTotalNumberOfMemorySegments(), buffers.size());
+
+			assertNull(fixedPool.requestBuffer());
+			assertNull(boundedPool.requestBuffer());
+			assertNull(nonFixedPool.requestBuffer());
 
 			// destroy all allocated ones
 			globalPool.destroyAllBufferPools();
 
 			// check the destroyed status
 			assertFalse(globalPool.isDestroyed());
-			assertTrue(lbp.isDestroyed());
+			assertTrue(fixedPool.isDestroyed());
+			assertTrue(boundedPool.isDestroyed());
+			assertTrue(nonFixedPool.isDestroyed());
 
 			assertEquals(0, globalPool.getNumberOfRegisteredBufferPools());
 
@@ -107,7 +137,23 @@ public class NetworkBufferPoolTest {
 
 			// can request no more buffers
 			try {
-				lbp.requestBuffer();
+				fixedPool.requestBuffer();
+				fail("Should fail with an IllegalStateException");
+			}
+			catch (IllegalStateException e) {
+				// yippie!
+			}
+
+			try {
+				boundedPool.requestBuffer();
+				fail("Should fail with an IllegalStateException");
+			}
+			catch (IllegalStateException e) {
+				// that's the way we like it, aha, aha
+			}
+
+			try {
+				nonFixedPool.requestBuffer();
 				fail("Should fail with an IllegalStateException");
 			}
 			catch (IllegalStateException e) {
@@ -115,7 +161,7 @@ public class NetworkBufferPoolTest {
 			}
 
 			// can create a new pool now
-			assertNotNull(globalPool.createBufferPool(8));
+			assertNotNull(globalPool.createBufferPool(10, Integer.MAX_VALUE));
 		}
 		catch (Exception e) {
 			e.printStackTrace();

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
index 27177c9..64f82f3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateConcurrentTest.java
@@ -60,7 +60,7 @@ public class InputGateConcurrentTest {
 		final SingleInputGate gate = new SingleInputGate(
 				"Test Task Name",
 				new JobID(),
-				new IntermediateDataSetID(),
+				new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
 				0, numChannels,
 				mock(TaskActions.class),
 				new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
@@ -95,7 +95,7 @@ public class InputGateConcurrentTest {
 		final SingleInputGate gate = new SingleInputGate(
 				"Test Task Name",
 				new JobID(),
-				new IntermediateDataSetID(),
+				new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
 				0,
 				numChannels,
 				mock(TaskActions.class),
@@ -144,7 +144,7 @@ public class InputGateConcurrentTest {
 		final SingleInputGate gate = new SingleInputGate(
 				"Test Task Name",
 				new JobID(),
-				new IntermediateDataSetID(),
+				new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
 				0,
 				numChannels,
 				mock(TaskActions.class),

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
index 7e1d792..f933840 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/InputGateFairnessTest.java
@@ -347,7 +347,8 @@ public class InputGateFairnessTest {
 				TaskActions taskActions,
 				TaskIOMetricGroup metrics) {
 
-			super(owningTaskName, jobId, consumedResultId, consumedSubpartitionIndex,
+			super(owningTaskName, jobId, consumedResultId, ResultPartitionType.PIPELINED,
+				consumedSubpartitionIndex,
 					numberOfInputChannels, taskActions, metrics);
 
 			try {

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
index 15ff2da..18c3038 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java
@@ -128,7 +128,7 @@ public class LocalInputChannelTest {
 
 			// Create a buffer pool for this partition
 			partition.registerBufferPool(
-				networkBuffers.createBufferPool(producerBufferPoolSize));
+				networkBuffers.createBufferPool(producerBufferPoolSize, producerBufferPoolSize));
 
 			// Create the producer
 			partitionProducers[i] = new TestPartitionProducer(
@@ -162,7 +162,7 @@ public class LocalInputChannelTest {
 						i,
 						parallelism,
 						numberOfBuffersPerChannel,
-						networkBuffers.createBufferPool(parallelism),
+						networkBuffers.createBufferPool(parallelism, parallelism),
 						partitionManager,
 						new TaskEventDispatcher(),
 						partitionIds)));
@@ -284,6 +284,7 @@ public class LocalInputChannelTest {
 			"test task name",
 			new JobID(),
 			new IntermediateDataSetID(),
+			ResultPartitionType.PIPELINED,
 			0,
 			1,
 			mock(TaskActions.class),
@@ -481,6 +482,7 @@ public class LocalInputChannelTest {
 					"Test Name",
 					new JobID(),
 					new IntermediateDataSetID(),
+					ResultPartitionType.PIPELINED,
 					subpartitionIndex,
 					numberOfInputChannels,
 					mock(TaskActions.class),

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index a25b8d5..2d1b4b2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -37,6 +37,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferRecycler;
 import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
 import org.apache.flink.runtime.io.network.util.TestTaskEvent;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
@@ -73,7 +74,13 @@ public class SingleInputGateTest {
 	public void testBasicGetNextLogic() throws Exception {
 		// Setup
 		final SingleInputGate inputGate = new SingleInputGate(
-			"Test Task Name", new JobID(), new IntermediateDataSetID(), 0, 2, mock(TaskActions.class), new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
+			"Test Task Name", new JobID(),
+			new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
+			0, 2,
+			mock(TaskActions.class),
+			new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
+
+		assertEquals(ResultPartitionType.PIPELINED, inputGate.getConsumedPartitionType());
 
 		final TestInputChannel[] inputChannels = new TestInputChannel[]{
 			new TestInputChannel(inputGate, 0),
@@ -127,7 +134,12 @@ public class SingleInputGateTest {
 		// Setup reader with one local and one unknown input channel
 		final IntermediateDataSetID resultId = new IntermediateDataSetID();
 
-		final SingleInputGate inputGate = new SingleInputGate("Test Task Name", new JobID(), resultId, 0, 2, mock(TaskActions.class), new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
+		final SingleInputGate inputGate = new SingleInputGate(
+				"Test Task Name", new JobID(),
+				resultId, ResultPartitionType.PIPELINED,
+				0, 2,
+				mock(TaskActions.class),
+				new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
 		final BufferPool bufferPool = mock(BufferPool.class);
 		when(bufferPool.getNumberOfRequiredMemorySegments()).thenReturn(2);
 
@@ -179,6 +191,7 @@ public class SingleInputGateTest {
 			"t1",
 			new JobID(),
 			new IntermediateDataSetID(),
+			ResultPartitionType.PIPELINED,
 			0,
 			1,
 			mock(TaskActions.class), new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
@@ -218,6 +231,7 @@ public class SingleInputGateTest {
 			"InputGate",
 			new JobID(),
 			new IntermediateDataSetID(),
+			ResultPartitionType.PIPELINED,
 			0,
 			1,
 			mock(TaskActions.class),
@@ -303,7 +317,9 @@ public class SingleInputGateTest {
 				partitionIds[2],
 				ResultPartitionLocation.createUnknown())};
 
-		InputGateDeploymentDescriptor gateDesc = new InputGateDeploymentDescriptor(new IntermediateDataSetID(), 0, channelDescs);
+		InputGateDeploymentDescriptor gateDesc =
+			new InputGateDeploymentDescriptor(new IntermediateDataSetID(),
+				ResultPartitionType.PIPELINED, 0, channelDescs);
 
 		int initialBackoff = 137;
 		int maxBackoff = 1001;
@@ -324,6 +340,8 @@ public class SingleInputGateTest {
 			mock(TaskActions.class),
 			new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
 
+		assertEquals(gateDesc.getConsumedPartitionType(), gate.getConsumedPartitionType());
+
 		Map<IntermediateResultPartitionID, InputChannel> channelMap = gate.getInputChannels();
 
 		assertEquals(3, channelMap.size());

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
index fe3b087..18ad490 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestSingleInputGate.java
@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.io.network.partition.consumer;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
 import org.apache.flink.runtime.operators.testutils.UnregisteredTaskMetricsGroup;
@@ -55,6 +56,7 @@ public class TestSingleInputGate {
 			"Test Task Name",
 			new JobID(),
 			new IntermediateDataSetID(),
+			ResultPartitionType.PIPELINED,
 			0,
 			numberOfInputChannels,
 			mock(TaskActions.class),

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
index c05df0a..bc1dd07 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.io.network.partition.consumer;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.runtime.operators.testutils.UnregisteredTaskMetricsGroup;
 import org.apache.flink.runtime.taskmanager.TaskActions;
@@ -42,8 +43,18 @@ public class UnionInputGateTest {
 	public void testBasicGetNextLogic() throws Exception {
 		// Setup
 		final String testTaskName = "Test Task";
-		final SingleInputGate ig1 = new SingleInputGate(testTaskName, new JobID(), new IntermediateDataSetID(), 0, 3, mock(TaskActions.class), new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
-		final SingleInputGate ig2 = new SingleInputGate(testTaskName, new JobID(), new IntermediateDataSetID(), 0, 5, mock(TaskActions.class), new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
+		final SingleInputGate ig1 = new SingleInputGate(
+			testTaskName, new JobID(),
+			new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
+			0, 3,
+			mock(TaskActions.class),
+			new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
+		final SingleInputGate ig2 = new SingleInputGate(
+			testTaskName, new JobID(),
+			new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
+			0, 5,
+			mock(TaskActions.class),
+			new UnregisteredTaskMetricsGroup.DummyTaskIOMetricGroup());
 
 		final UnionInputGate union = new UnionInputGate(new SingleInputGate[]{ig1, ig2});
 

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
index e26e176..cecfe6a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java
@@ -113,7 +113,8 @@ public class TaskManagerComponentsStartupShutdownTest {
 				false); // exit-jvm-on-fatal-error
 
 			final NetworkEnvironmentConfiguration netConf = new NetworkEnvironmentConfiguration(
-					32, BUFFER_SIZE, MemoryType.HEAP, IOManager.IOMode.SYNC, 0, 0, null);
+					32, BUFFER_SIZE, MemoryType.HEAP, IOManager.IOMode.SYNC,
+					0, 0, 2, 8, null);
 
 			ResourceID taskManagerId = ResourceID.generate();
 			
@@ -130,7 +131,9 @@ public class TaskManagerComponentsStartupShutdownTest {
 				null,
 				netConf.ioMode(),
 				netConf.partitionRequestInitialBackoff(),
-				netConf.partitionRequestMaxBackoff());
+				netConf.partitionRequestMaxBackoff(),
+				netConf.networkBuffersPerChannel(),
+				netConf.extraNetworkBuffersPerGate());
 
 			network.start();
 

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
index 356d693..0b39e20 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -69,6 +69,7 @@ import org.apache.flink.runtime.messages.TaskMessages;
 import org.apache.flink.runtime.messages.TaskMessages.CancelTask;
 import org.apache.flink.runtime.messages.TaskMessages.StopTask;
 import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
+import org.apache.flink.runtime.taskexecutor.TaskManagerServicesConfiguration;
 import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
 import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
 import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -91,6 +92,7 @@ import scala.concurrent.duration.FiniteDuration;
 import scala.util.Failure;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.URL;
 import java.util.ArrayList;
@@ -637,7 +639,7 @@ public class TaskManagerTest extends TestLogger {
 
 				InputGateDeploymentDescriptor ircdd =
 						new InputGateDeploymentDescriptor(
-								new IntermediateDataSetID(),
+								new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
 								0, new InputChannelDeploymentDescriptor[]{
 										new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), ResultPartitionLocation.createLocal())
 								}
@@ -782,7 +784,7 @@ public class TaskManagerTest extends TestLogger {
 
 				InputGateDeploymentDescriptor ircdd =
 						new InputGateDeploymentDescriptor(
-								new IntermediateDataSetID(),
+								new IntermediateDataSetID(), ResultPartitionType.PIPELINED,
 								0, new InputChannelDeploymentDescriptor[]{
 										new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), ResultPartitionLocation.createLocal())
 								}
@@ -936,7 +938,7 @@ public class TaskManagerTest extends TestLogger {
 								new InputChannelDeploymentDescriptor(partitionId, loc)};
 
 				final InputGateDeploymentDescriptor igdd =
-						new InputGateDeploymentDescriptor(resultId, 0, icdd);
+						new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, icdd);
 
 				final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor(
 						jid, "TestJob", vid, eid,
@@ -978,6 +980,25 @@ public class TaskManagerTest extends TestLogger {
 		}};
 	}
 
+	@Test
+	public void testTaskManagerServicesConfiguration() throws Exception {
+
+		// set some non-default values
+		final Configuration config = new Configuration();
+		config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
+		config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
+		config.setInteger(TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL, 10);
+		config.setInteger(TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100);
+
+		TaskManagerServicesConfiguration tmConfig =
+			TaskManagerServicesConfiguration.fromConfiguration(config, InetAddress.getByName("localhost"), true);
+
+		assertEquals(tmConfig.getNetworkConfig().partitionRequestInitialBackoff(), 100);
+		assertEquals(tmConfig.getNetworkConfig().partitionRequestMaxBackoff(), 200);
+		assertEquals(tmConfig.getNetworkConfig().networkBuffersPerChannel(), 10);
+		assertEquals(tmConfig.getNetworkConfig().extraNetworkBuffersPerGate(), 100);
+	}
+
 	/**
 	 *  Tests that repeated local {@link PartitionNotFoundException}s ultimately fail the receiver.
 	 */
@@ -1031,7 +1052,7 @@ public class TaskManagerTest extends TestLogger {
 								new InputChannelDeploymentDescriptor(partitionId, loc)};
 
 				final InputGateDeploymentDescriptor igdd =
-						new InputGateDeploymentDescriptor(resultId, 0, icdd);
+						new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, icdd);
 
 				final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor(
 						jid, "TestJob", vid, eid,

http://git-wip-us.apache.org/repos/asf/flink/blob/11e2aa6d/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index 003eff9..c18b527 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -402,17 +402,17 @@ public class StreamingJobGraphGenerator {
 			jobEdge = downStreamVertex.connectNewDataSetAsInput(
 				headVertex,
 				DistributionPattern.POINTWISE,
-				ResultPartitionType.PIPELINED);
+				ResultPartitionType.PIPELINED_BOUNDED);
 		} else if (partitioner instanceof RescalePartitioner){
 			jobEdge = downStreamVertex.connectNewDataSetAsInput(
 				headVertex,
 				DistributionPattern.POINTWISE,
-				ResultPartitionType.PIPELINED);
+				ResultPartitionType.PIPELINED_BOUNDED);
 		} else {
 			jobEdge = downStreamVertex.connectNewDataSetAsInput(
 					headVertex,
 					DistributionPattern.ALL_TO_ALL,
-					ResultPartitionType.PIPELINED);
+					ResultPartitionType.PIPELINED_BOUNDED);
 		}
 		// set strategy name so that web interface can show it.
 		jobEdge.setShipStrategyName(partitioner.toString());


Mime
View raw message