Github user NicoK commented on a diff in the pull request:
https://github.com/apache/flink/pull/4509#discussion_r152860980
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java
---
@@ -506,36 +540,22 @@ public Void call() throws Exception {
@Test
public void testConcurrentOnSenderBacklogAndRecycle() throws Exception {
// Setup
- final NetworkBufferPool networkBufferPool = new
NetworkBufferPool(256, 32, MemoryType.HEAP);
- final ExecutorService executor =
Executors.newFixedThreadPool(2);
+ final NetworkBufferPool networkBufferPool = new
NetworkBufferPool(248, 32, MemoryType.HEAP);
+ final int numExclusiveSegments = 120;
+ final int numFloatingBuffers = 128;
+ final int backlog = 128;
+
+ final ExecutorService executor =
Executors.newFixedThreadPool(3);
+
final SingleInputGate inputGate = createSingleInputGate();
final RemoteInputChannel inputChannel =
createRemoteInputChannel(inputGate);
inputGate.setInputChannel(inputChannel.partitionId.getPartitionId(),
inputChannel);
try {
- final int numFloatingBuffers = 128;
- final int numExclusiveSegments = 2;
final BufferPool bufferPool =
networkBufferPool.createBufferPool(numFloatingBuffers, numFloatingBuffers);
inputGate.setBufferPool(bufferPool);
inputGate.assignExclusiveSegments(networkBufferPool,
numExclusiveSegments);
- // Exhaust all the floating buffers
- final List<Buffer> floatingBuffers = new
ArrayList<>(numFloatingBuffers);
- for (int i = 0; i < numFloatingBuffers; i++) {
- Buffer buffer = bufferPool.requestBuffer();
- assertNotNull(buffer);
- floatingBuffers.add(buffer);
- }
-
- // Exhaust all the exclusive buffers
- final List<Buffer> exclusiveBuffers = new
ArrayList<>(numExclusiveSegments);
- for (int i = 0; i < numExclusiveSegments; i++) {
- Buffer buffer = inputChannel.requestBuffer();
- assertNotNull(buffer);
- exclusiveBuffers.add(buffer);
- }
-
- final int backlog = 128;
- final Callable<Void> requestBufferTask = new
Callable<Void>() {
+ final Callable requestBufferTask = new Callable<Void>()
{
--- End diff --
please keep `Callable<Void>` (or replace by a lambda)
---
