You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/01/18 16:06:13 UTC
[1/3] flink git commit: [hotfix][network][tests] remove Mockito mocks
in RecordWriterTest
Repository: flink
Updated Branches:
refs/heads/master 665347cf8 -> 03c797a1d
[hotfix][network][tests] remove Mockito mocks in RecordWriterTest
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/905d98e3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/905d98e3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/905d98e3
Branch: refs/heads/master
Commit: 905d98e37c015d24d097113b6e6e8194bd88cbd8
Parents: 7583365
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Thu Jan 11 14:35:30 2018 +0100
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Thu Jan 18 17:05:00 2018 +0100
----------------------------------------------------------------------
.../io/network/api/writer/RecordWriterTest.java | 185 ++++++++++++-------
1 file changed, 119 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/905d98e3/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 9e281d8..eac1f8b 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
@@ -38,6 +38,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferRecycler;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
import org.apache.flink.runtime.io.network.util.TestTaskEvent;
@@ -74,6 +75,7 @@ import static org.mockito.Matchers.anyInt;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -126,7 +128,7 @@ public class RecordWriterTest {
BufferProvider bufferProvider = mock(BufferProvider.class);
when(bufferProvider.requestBufferBuilderBlocking()).thenAnswer(request);
- ResultPartitionWriter partitionWriter = createResultPartitionWriter(bufferProvider);
+ ResultPartitionWriter partitionWriter = spy(new RecyclingPartitionWriter(bufferProvider));
final RecordWriter<IntValue> recordWriter = new RecordWriter<IntValue>(partitionWriter);
@@ -306,8 +308,8 @@ public class RecordWriterTest {
@Test
public void testSerializerClearedAfterClearBuffers() throws Exception {
- ResultPartitionWriter partitionWriter = createResultPartitionWriter(
- new TestPooledBufferProvider(1, 16));
+ ResultPartitionWriter partitionWriter =
+ spy(new RecyclingPartitionWriter(new TestPooledBufferProvider(1, 16)));
RecordWriter<IntValue> recordWriter = new RecordWriter<IntValue>(partitionWriter);
@@ -332,14 +334,14 @@ public class RecordWriterTest {
int bufferSize = 32;
@SuppressWarnings("unchecked")
- Queue<BufferOrEvent>[] queues = new Queue[numChannels];
+ Queue<Buffer>[] queues = new Queue[numChannels];
for (int i = 0; i < numChannels; i++) {
queues[i] = new ArrayDeque<>();
}
TestPooledBufferProvider bufferProvider = new TestPooledBufferProvider(Integer.MAX_VALUE, bufferSize);
- ResultPartitionWriter partitionWriter = createCollectingPartitionWriter(queues, bufferProvider);
+ ResultPartitionWriter partitionWriter = new CollectingPartitionWriter(queues, bufferProvider);
RecordWriter<ByteArrayIO> writer = new RecordWriter<>(partitionWriter, new RoundRobin<ByteArrayIO>());
CheckpointBarrier barrier = new CheckpointBarrier(Integer.MAX_VALUE + 919192L, Integer.MAX_VALUE + 18828228L, CheckpointOptions.forCheckpoint());
@@ -348,11 +350,12 @@ public class RecordWriterTest {
assertEquals(0, bufferProvider.getNumberOfCreatedBuffers());
- for (Queue<BufferOrEvent> queue : queues) {
- assertEquals(1, queue.size());
- BufferOrEvent boe = queue.remove();
+ for (int i = 0; i < numChannels; i++) {
+ assertEquals(1, queues[i].size());
+ BufferOrEvent boe = parseBuffer(queues[i].remove(), i);
assertTrue(boe.isEvent());
assertEquals(barrier, boe.getEvent());
+ assertEquals(0, queues[i].size());
}
}
@@ -368,14 +371,14 @@ public class RecordWriterTest {
int lenBytes = 4; // serialized length
@SuppressWarnings("unchecked")
- Queue<BufferOrEvent>[] queues = new Queue[numChannels];
+ Queue<Buffer>[] queues = new Queue[numChannels];
for (int i = 0; i < numChannels; i++) {
queues[i] = new ArrayDeque<>();
}
TestPooledBufferProvider bufferProvider = new TestPooledBufferProvider(Integer.MAX_VALUE, bufferSize);
- ResultPartitionWriter partitionWriter = createCollectingPartitionWriter(queues, bufferProvider);
+ ResultPartitionWriter partitionWriter = new CollectingPartitionWriter(queues, bufferProvider);
RecordWriter<ByteArrayIO> writer = new RecordWriter<>(partitionWriter, new RoundRobin<ByteArrayIO>());
CheckpointBarrier barrier = new CheckpointBarrier(Integer.MAX_VALUE + 1292L, Integer.MAX_VALUE + 199L, CheckpointOptions.forCheckpoint());
@@ -408,10 +411,22 @@ public class RecordWriterTest {
assertEquals(4, bufferProvider.getNumberOfCreatedBuffers());
+ BufferOrEvent boe;
assertEquals(2, queues[0].size()); // 1 buffer + 1 event
+ assertTrue(parseBuffer(queues[0].remove(), 0).isBuffer());
assertEquals(3, queues[1].size()); // 2 buffers + 1 event
+ assertTrue(parseBuffer(queues[1].remove(), 1).isBuffer());
+ assertTrue(parseBuffer(queues[1].remove(), 1).isBuffer());
assertEquals(2, queues[2].size()); // 1 buffer + 1 event
+ assertTrue(parseBuffer(queues[2].remove(), 2).isBuffer());
assertEquals(1, queues[3].size()); // 0 buffers + 1 event
+
+ // every queue's last element should be the event
+ for (int i = 0; i < numChannels; i++) {
+ boe = parseBuffer(queues[i].remove(), i);
+ assertTrue(boe.isEvent());
+ assertEquals(barrier, boe.getEvent());
+ }
}
/**
@@ -430,11 +445,10 @@ public class RecordWriterTest {
.toReturn(buffer);
@SuppressWarnings("unchecked")
- ArrayDeque<BufferOrEvent>[] queues =
- new ArrayDeque[]{new ArrayDeque(), new ArrayDeque()};
+ ArrayDeque<Buffer>[] queues = new ArrayDeque[] { new ArrayDeque(), new ArrayDeque() };
ResultPartitionWriter partition =
- createCollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE));
+ new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE));
RecordWriter<?> writer = new RecordWriter<>(partition);
writer.broadcastEvent(EndOfPartitionEvent.INSTANCE);
@@ -443,6 +457,11 @@ public class RecordWriterTest {
assertEquals(1, queues[0].size());
assertEquals(1, queues[1].size());
+ // process all collected events (recycles the buffer)
+ for (int i = 0; i < queues.length; i++) {
+ assertTrue(parseBuffer(queues[i].remove(), i).isEvent());
+ }
+
assertTrue(buffer.isRecycled());
}
@@ -450,45 +469,6 @@ public class RecordWriterTest {
// Helpers
// ---------------------------------------------------------------------------------------------
- /**
- * Creates a mock partition writer that collects the added buffers/events.
- *
- * <p>This much mocking should not be necessary with better designed
- * interfaces. Refactoring this will take too much time now though, hence
- * the mocking. Ideally, we will refactor all of this mess in order to make
- * our lives easier and test it better.
- */
- private ResultPartitionWriter createCollectingPartitionWriter(
- final Queue<BufferOrEvent>[] queues,
- BufferProvider bufferProvider) throws IOException {
-
- int numChannels = queues.length;
-
- ResultPartitionWriter partitionWriter = mock(ResultPartitionWriter.class);
- when(partitionWriter.getBufferProvider()).thenReturn(checkNotNull(bufferProvider));
- when(partitionWriter.getNumberOfSubpartitions()).thenReturn(numChannels);
-
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
- Buffer buffer = (Buffer) invocationOnMock.getArguments()[0];
- if (buffer.isBuffer()) {
- Integer targetChannel = (Integer) invocationOnMock.getArguments()[1];
- queues[targetChannel].add(new BufferOrEvent(buffer, targetChannel));
- } else {
- // is event:
- AbstractEvent event = EventSerializer.fromBuffer(buffer, getClass().getClassLoader());
- buffer.recycleBuffer(); // the buffer is not needed anymore
- Integer targetChannel = (Integer) invocationOnMock.getArguments()[1];
- queues[targetChannel].add(new BufferOrEvent(event, targetChannel));
- }
- return null;
- }
- }).when(partitionWriter).writeBuffer(any(Buffer.class), anyInt());
-
- return partitionWriter;
- }
-
private BufferProvider createBufferProvider(final int bufferSize)
throws IOException, InterruptedException {
@@ -519,24 +499,97 @@ public class RecordWriterTest {
return bufferProvider;
}
- private ResultPartitionWriter createResultPartitionWriter(BufferProvider bufferProvider)
- throws IOException {
+ /**
+ * Partition writer that collects the added buffers/events in multiple queue.
+ */
+ private static class CollectingPartitionWriter implements ResultPartitionWriter {
+ private final Queue<Buffer>[] queues;
+ private final BufferProvider bufferProvider;
+ private final ResultPartitionID partitionId = new ResultPartitionID();
+
+ /**
+ * Create the partition writer.
+ *
+ * @param queues one queue per outgoing channel
+ * @param bufferProvider buffer provider
+ */
+ private CollectingPartitionWriter(Queue<Buffer>[] queues, BufferProvider bufferProvider) {
+ this.queues = queues;
+ this.bufferProvider = bufferProvider;
+ }
- ResultPartitionWriter partitionWriter = mock(ResultPartitionWriter.class);
- when(partitionWriter.getBufferProvider()).thenReturn(checkNotNull(bufferProvider));
- when(partitionWriter.getNumberOfSubpartitions()).thenReturn(1);
+ @Override
+ public BufferProvider getBufferProvider() {
+ return bufferProvider;
+ }
- // Recycle each written buffer.
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- ((Buffer) invocation.getArguments()[0]).recycleBuffer();
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return partitionId;
+ }
- return null;
- }
- }).when(partitionWriter).writeBuffer(any(Buffer.class), anyInt());
+ @Override
+ public int getNumberOfSubpartitions() {
+ return queues.length;
+ }
- return partitionWriter;
+ @Override
+ public int getNumTargetKeyGroups() {
+ return 1;
+ }
+
+ @Override
+ public void writeBuffer(Buffer buffer, int targetChannel) throws IOException {
+ queues[targetChannel].add(buffer);
+ }
+ }
+
+ private static BufferOrEvent parseBuffer(Buffer buffer, int targetChannel) throws IOException {
+ if (buffer.isBuffer()) {
+ return new BufferOrEvent(buffer, targetChannel);
+ } else {
+ // is event:
+ AbstractEvent event = EventSerializer.fromBuffer(buffer, RecordWriterTest.class.getClassLoader());
+ buffer.recycleBuffer(); // the buffer is not needed anymore
+ return new BufferOrEvent(event, targetChannel);
+ }
+ }
+
+ /**
+ * Partition writer that recycles all received buffers and does no further processing.
+ */
+ private static class RecyclingPartitionWriter implements ResultPartitionWriter {
+ private final BufferProvider bufferProvider;
+ private final ResultPartitionID partitionId = new ResultPartitionID();
+
+ private RecyclingPartitionWriter(BufferProvider bufferProvider) {
+ this.bufferProvider = bufferProvider;
+ }
+
+ @Override
+ public BufferProvider getBufferProvider() {
+ return bufferProvider;
+ }
+
+ @Override
+ public ResultPartitionID getPartitionId() {
+ return partitionId;
+ }
+
+ @Override
+ public int getNumberOfSubpartitions() {
+ return 1;
+ }
+
+ @Override
+ public int getNumTargetKeyGroups() {
+ return 1;
+ }
+
+ @Override
+ public void writeBuffer(Buffer buffer, int targetChannel) throws IOException {
+ buffer.recycleBuffer();
+ }
}
private static class ByteArrayIO implements IOReadableWritable {
[2/3] flink git commit: [hotfix][tests] separate tests inside
SlotCountExceedingParallelismTest
Posted by sr...@apache.org.
[hotfix][tests] separate tests inside SlotCountExceedingParallelismTest
This way, in case of failures, we may have better pointers on what test code is
currently executed.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/75833655
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/75833655
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/75833655
Branch: refs/heads/master
Commit: 758336553248d2414a88b071ebd5748c1c32d282
Parents: 665347c
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Tue Aug 29 11:49:47 2017 +0200
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Thu Jan 18 17:05:00 2018 +0100
----------------------------------------------------------------------
.../SlotCountExceedingParallelismTest.java | 18 ++++++++++++------
1 file changed, 12 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/75833655/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
index 4ce83cc..a881233 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
@@ -45,6 +45,8 @@ public class SlotCountExceedingParallelismTest extends TestLogger {
private static final int NUMBER_OF_SLOTS_PER_TM = 2;
private static final int PARALLELISM = NUMBER_OF_TMS * NUMBER_OF_SLOTS_PER_TM;
+ public static final String JOB_NAME = "SlotCountExceedingParallelismTest (no slot sharing, blocking results)";
+
private static TestingCluster flink;
@BeforeClass
@@ -63,19 +65,23 @@ public class SlotCountExceedingParallelismTest extends TestLogger {
}
@Test
- public void testNoSlotSharingAndBlockingResult() throws Exception {
- final String jobName = "SlotCountExceedingParallelismTest (no slot sharing, blocking results)";
-
+ public void testNoSlotSharingAndBlockingResultSender() throws Exception {
// Sender with higher parallelism than available slots
- JobGraph jobGraph = createTestJobGraph(jobName, PARALLELISM * 2, PARALLELISM);
+ JobGraph jobGraph = createTestJobGraph(JOB_NAME, PARALLELISM * 2, PARALLELISM);
submitJobGraphAndWait(jobGraph);
+ }
+ @Test
+ public void testNoSlotSharingAndBlockingResultReceiver() throws Exception {
// Receiver with higher parallelism than available slots
- jobGraph = createTestJobGraph(jobName, PARALLELISM, PARALLELISM * 2);
+ JobGraph jobGraph = createTestJobGraph(JOB_NAME, PARALLELISM, PARALLELISM * 2);
submitJobGraphAndWait(jobGraph);
+ }
+ @Test
+ public void testNoSlotSharingAndBlockingResultBoth() throws Exception {
// Both sender and receiver with higher parallelism than available slots
- jobGraph = createTestJobGraph(jobName, PARALLELISM * 2, PARALLELISM * 2);
+ JobGraph jobGraph = createTestJobGraph(JOB_NAME, PARALLELISM * 2, PARALLELISM * 2);
submitJobGraphAndWait(jobGraph);
}
[3/3] flink git commit: [FLINK-7518][network] pass our own
NetworkBuffer to netty
Posted by sr...@apache.org.
[FLINK-7518][network] pass our own NetworkBuffer to netty
This is using a composite buffer to assemble header+content and avoids an
unnecessary buffer copy from our (Network)Buffer class backed by a MemorySegment
to Netty's ByteBuf class.
This closes #4615.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/03c797a1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/03c797a1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/03c797a1
Branch: refs/heads/master
Commit: 03c797a1d279a6c09c988755f94566a3d46cbb17
Parents: 905d98e
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Fri Aug 25 12:13:54 2017 +0200
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Thu Jan 18 17:05:32 2018 +0100
----------------------------------------------------------------------
.../io/network/api/writer/RecordWriter.java | 3 +-
.../flink/runtime/io/network/buffer/Buffer.java | 6 +
.../io/network/buffer/NetworkBuffer.java | 5 +
.../buffer/ReadOnlySlicedNetworkBuffer.java | 4 +
.../network/netty/CreditBasedClientHandler.java | 4 +-
.../runtime/io/network/netty/NettyMessage.java | 139 ++++++++++++-------
.../netty/PartitionRequestClientHandler.java | 12 +-
.../BufferFileWriterFileSegmentReaderTest.java | 2 +-
.../io/network/api/writer/RecordWriterTest.java | 58 ++++++++
.../buffer/ReadOnlySlicedBufferTest.java | 2 +-
.../netty/NettyMessageSerializationTest.java | 83 +++++------
.../PartitionRequestClientHandlerTest.java | 3 +-
.../netty/PartitionRequestQueueTest.java | 119 ++++++++++++++++
13 files changed, 334 insertions(+), 106 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
index de1cac6..0bf12c4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
@@ -155,8 +155,7 @@ public class RecordWriter<T extends IOReadableWritable> {
}
// retain the buffer so that it can be recycled by each channel of targetPartition
- eventBuffer.retainBuffer();
- targetPartition.writeBuffer(eventBuffer, targetChannel);
+ targetPartition.writeBuffer(eventBuffer.readOnlySlice().retainBuffer(), targetChannel);
}
}
} finally {
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
index 3c9933e..c649262 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.io.network.buffer;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator;
import java.nio.ByteBuffer;
@@ -210,4 +211,9 @@ public interface Buffer {
* @param allocator netty buffer allocator
*/
void setAllocator(ByteBufAllocator allocator);
+
+ /**
+ * @return self as ByteBuf implementation.
+ */
+ ByteBuf asByteBuf();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
index 4ca50db..0f3fa0e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/NetworkBuffer.java
@@ -529,4 +529,9 @@ public class NetworkBuffer extends AbstractReferenceCountedByteBuf implements Bu
.append(')');
return buf.toString();
}
+
+ @Override
+ public ByteBuf asByteBuf() {
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
index f268314..d5f81d0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedNetworkBuffer.java
@@ -196,4 +196,8 @@ public final class ReadOnlySlicedNetworkBuffer extends ReadOnlyByteBuf implement
((Buffer) unwrap()).setAllocator(allocator);
}
+ @Override
+ public ByteBuf asByteBuf() {
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
index fdcc179..8f171b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/CreditBasedClientHandler.java
@@ -305,9 +305,9 @@ class CreditBasedClientHandler extends ChannelInboundHandlerAdapter {
return;
}
- NetworkBuffer buffer = (NetworkBuffer) inputChannel.requestBuffer();
+ Buffer buffer = inputChannel.requestBuffer();
if (buffer != null) {
- nettyBuffer.readBytes(buffer, size);
+ nettyBuffer.readBytes(buffer.asByteBuf(), size);
inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, bufferOrEvent.backlog);
} else if (inputChannel.isReleased()) {
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index f05febc..696c74e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
@@ -26,11 +26,13 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufInputStream;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufOutputStream;
+import org.apache.flink.shaded.netty4.io.netty.buffer.CompositeByteBuf;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOutboundHandlerAdapter;
@@ -63,7 +65,7 @@ public abstract class NettyMessage {
// constructor in order to work with the generic deserializer.
// ------------------------------------------------------------------------
- static final int HEADER_LENGTH = 4 + 4 + 1; // frame length (4), magic number (4), msg ID (1)
+ static final int FRAME_HEADER_LENGTH = 4 + 4 + 1; // frame length (4), magic number (4), msg ID (1)
static final int MAGIC_NUMBER = 0xBADC0FFE;
@@ -94,31 +96,63 @@ public abstract class NettyMessage {
* Allocates a new (header and contents) buffer and adds some header information for the frame
* decoder.
*
- * <p>If the <tt>length</tt> is unknown, you must write the actual length after adding the
- * contents as an integer to position <tt>0</tt>!
+ * <p>If the <tt>contentLength</tt> is unknown, you must write the actual length after adding
+ * the contents as an integer to position <tt>0</tt>!
*
* @param allocator
* byte buffer allocator to use
* @param id
* {@link NettyMessage} subclass ID
- * @param length
+ * @param contentLength
* content length (or <tt>-1</tt> if unknown)
*
* @return a newly allocated direct buffer with header data written for {@link
* NettyMessageDecoder}
*/
- private static ByteBuf allocateBuffer(ByteBufAllocator allocator, byte id, int length) {
- checkArgument(length <= Integer.MAX_VALUE - HEADER_LENGTH);
+ private static ByteBuf allocateBuffer(ByteBufAllocator allocator, byte id, int contentLength) {
+ return allocateBuffer(allocator, id, 0, contentLength, true);
+ }
+
+ /**
+ * Allocates a new buffer and adds some header information for the frame decoder.
+ *
+ * <p>If the <tt>contentLength</tt> is unknown, you must write the actual length after adding
+ * the contents as an integer to position <tt>0</tt>!
+ *
+ * @param allocator
+ * byte buffer allocator to use
+ * @param id
+ * {@link NettyMessage} subclass ID
+ * @param messageHeaderLength
+ * additional header length that should be part of the allocated buffer and is written
+ * outside of this method
+ * @param contentLength
+ * content length (or <tt>-1</tt> if unknown)
+ * @param allocateForContent
+ * whether to make room for the actual content in the buffer (<tt>true</tt>) or whether to
+ * only return a buffer with the header information (<tt>false</tt>)
+ *
+ * @return a newly allocated direct buffer with header data written for {@link
+ * NettyMessageDecoder}
+ */
+ private static ByteBuf allocateBuffer(
+ ByteBufAllocator allocator,
+ byte id,
+ int messageHeaderLength,
+ int contentLength,
+ boolean allocateForContent) {
+ checkArgument(contentLength <= Integer.MAX_VALUE - FRAME_HEADER_LENGTH);
final ByteBuf buffer;
- if (length != -1) {
- buffer = allocator.directBuffer(HEADER_LENGTH + length);
+ if (!allocateForContent) {
+ buffer = allocator.directBuffer(FRAME_HEADER_LENGTH + messageHeaderLength);
+ } else if (contentLength != -1) {
+ buffer = allocator.directBuffer(FRAME_HEADER_LENGTH + messageHeaderLength + contentLength);
} else {
- // content length unknown -> start with the default initial size (rather than HEADER_LENGTH only):
+ // content length unknown -> start with the default initial size (rather than FRAME_HEADER_LENGTH only):
buffer = allocator.directBuffer();
}
-
- buffer.writeInt(HEADER_LENGTH + length);
+ buffer.writeInt(FRAME_HEADER_LENGTH + messageHeaderLength + contentLength); // may be updated later, e.g. if contentLength == -1
buffer.writeInt(MAGIC_NUMBER);
buffer.writeByte(id);
@@ -217,8 +251,7 @@ public abstract class NettyMessage {
private static final byte ID = 0;
- @Nullable
- final Buffer buffer;
+ final ByteBuf buffer;
final InputChannelID receiverId;
@@ -226,31 +259,27 @@ public abstract class NettyMessage {
final int backlog;
- // ---- Deserialization -----------------------------------------------
-
final boolean isBuffer;
- @Nullable
- ByteBuf retainedSlice;
-
private BufferResponse(
- ByteBuf retainedSlice, boolean isBuffer, int sequenceNumber,
+ ByteBuf buffer,
+ boolean isBuffer,
+ int sequenceNumber,
InputChannelID receiverId,
int backlog) {
- // When deserializing we first have to request a buffer from the respective buffer
- // provider (at the handler) and copy the buffer from Netty's space to ours. Only
- // retainedSlice is set in this case.
- this.buffer = null;
- this.retainedSlice = checkNotNull(retainedSlice);
+ this.buffer = checkNotNull(buffer);
this.isBuffer = isBuffer;
this.sequenceNumber = sequenceNumber;
this.receiverId = checkNotNull(receiverId);
this.backlog = backlog;
}
- BufferResponse(Buffer buffer, int sequenceNumber, InputChannelID receiverId, int backlog) {
- this.buffer = checkNotNull(buffer);
- this.retainedSlice = null;
+ BufferResponse(
+ Buffer buffer,
+ int sequenceNumber,
+ InputChannelID receiverId,
+ int backlog) {
+ this.buffer = checkNotNull(buffer).asByteBuf();
this.isBuffer = buffer.isBuffer();
this.sequenceNumber = sequenceNumber;
this.receiverId = checkNotNull(receiverId);
@@ -262,14 +291,11 @@ public abstract class NettyMessage {
}
ByteBuf getNettyBuffer() {
- return retainedSlice;
+ return buffer;
}
void releaseBuffer() {
- if (retainedSlice != null) {
- retainedSlice.release();
- retainedSlice = null;
- }
+ buffer.release();
}
// --------------------------------------------------------------------
@@ -278,34 +304,40 @@ public abstract class NettyMessage {
@Override
ByteBuf write(ByteBufAllocator allocator) throws IOException {
- checkNotNull(buffer, "No buffer instance to serialize.");
-
+ // receiver ID (16), sequence number (4), backlog (4), isBuffer (1), buffer size (4)
+ final int messageHeaderLength = 16 + 4 + 4 + 1 + 4;
- ByteBuf result = null;
+ ByteBuf headerBuf = null;
try {
- ByteBuffer nioBufferReadable = buffer.getNioBufferReadable();
- int length = 16 + 4 + 4 + 1 + 4 + nioBufferReadable.remaining();
-
- result = allocateBuffer(allocator, ID, length);
-
- receiverId.writeTo(result);
- result.writeInt(sequenceNumber);
- result.writeInt(backlog);
- result.writeBoolean(buffer.isBuffer());
- result.writeInt(nioBufferReadable.remaining());
- result.writeBytes(nioBufferReadable);
+ if (buffer instanceof Buffer) {
+ // in order to forward the buffer to netty, it needs an allocator set
+ ((Buffer) buffer).setAllocator(allocator);
+ }
- return result;
+ // only allocate header buffer - we will combine it with the data buffer below
+ headerBuf = allocateBuffer(allocator, ID, messageHeaderLength, buffer.readableBytes(), false);
+
+ receiverId.writeTo(headerBuf);
+ headerBuf.writeInt(sequenceNumber);
+ headerBuf.writeInt(backlog);
+ headerBuf.writeBoolean(isBuffer);
+ headerBuf.writeInt(buffer.readableBytes());
+
+ CompositeByteBuf composityBuf = allocator.compositeDirectBuffer();
+ composityBuf.addComponent(headerBuf);
+ composityBuf.addComponent(buffer);
+ // update writer index since we have data written to the components:
+ composityBuf.writerIndex(headerBuf.writerIndex() + buffer.writerIndex());
+ return composityBuf;
}
catch (Throwable t) {
- if (result != null) {
- result.release();
+ if (headerBuf != null) {
+ headerBuf.release();
}
+ buffer.release();
- throw new IOException(t);
- }
- finally {
- buffer.recycleBuffer();
+ ExceptionUtils.rethrowIOException(t);
+ return null; // silence the compiler
}
}
@@ -317,7 +349,6 @@ public abstract class NettyMessage {
int size = buffer.readInt();
ByteBuf retainedSlice = buffer.readSlice(size).retain();
-
return new BufferResponse(retainedSlice, isBuffer, sequenceNumber, receiverId, backlog);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
index ba7bf22..9468729 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandler.java
@@ -294,10 +294,10 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
}
while (true) {
- NetworkBuffer buffer = (NetworkBuffer) bufferProvider.requestBuffer();
+ Buffer buffer = bufferProvider.requestBuffer();
if (buffer != null) {
- nettyBuffer.readBytes(buffer, receivedSize);
+ nettyBuffer.readBytes(buffer.asByteBuf(), receivedSize);
inputChannel.onBuffer(buffer, bufferOrEvent.sequenceNumber, -1);
@@ -369,7 +369,7 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
*/
private class BufferListenerTask implements BufferListener, Runnable {
- private final AtomicReference<NetworkBuffer> availableBuffer = new AtomicReference<>();
+ private final AtomicReference<Buffer> availableBuffer = new AtomicReference<>();
private NettyMessage.BufferResponse stagedBufferResponse;
@@ -414,7 +414,7 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
boolean success = false;
try {
- if (availableBuffer.compareAndSet(null, (NetworkBuffer) buffer)) {
+ if (availableBuffer.compareAndSet(null, buffer)) {
ctx.channel().eventLoop().execute(this);
success = true;
@@ -447,7 +447,7 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
public void run() {
boolean success = false;
- NetworkBuffer buffer = null;
+ Buffer buffer = null;
try {
if ((buffer = availableBuffer.getAndSet(null)) == null) {
@@ -455,7 +455,7 @@ class PartitionRequestClientHandler extends ChannelInboundHandlerAdapter {
}
ByteBuf nettyBuffer = stagedBufferResponse.getNettyBuffer();
- nettyBuffer.readBytes(buffer, nettyBuffer.readableBytes());
+ nettyBuffer.readBytes(buffer.asByteBuf(), nettyBuffer.readableBytes());
stagedBufferResponse.releaseBuffer();
RemoteInputChannel inputChannel = inputChannels.get(stagedBufferResponse.receiverId);
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriterFileSegmentReaderTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriterFileSegmentReaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriterFileSegmentReaderTest.java
index 2b52184..1aaefcb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriterFileSegmentReaderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/disk/iomanager/BufferFileWriterFileSegmentReaderTest.java
@@ -154,7 +154,7 @@ public class BufferFileWriterFileSegmentReaderTest {
fileSegment.getFileChannel().read(buffer, fileSegment.getPosition());
- NetworkBuffer buffer1 = new NetworkBuffer(MemorySegmentFactory.wrap(buffer.array()), BUFFER_RECYCLER);
+ Buffer buffer1 = new NetworkBuffer(MemorySegmentFactory.wrap(buffer.array()), BUFFER_RECYCLER);
buffer1.setSize(fileSegment.getLength());
currentNumber = verifyBufferFilledWithAscendingNumbers(buffer1, currentNumber);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/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 eac1f8b..9447e57 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
@@ -465,6 +465,64 @@ public class RecordWriterTest {
assertTrue(buffer.isRecycled());
}
+ /**
+ * Tests that broadcasted events' buffers are independent (in their (reader) indices) once they
+ * are put into the queue for Netty when broadcasting events to multiple channels.
+ */
+ @Test
+ public void testBroadcastEventBufferIndependence() throws Exception {
+ @SuppressWarnings("unchecked")
+ ArrayDeque<Buffer>[] queues =
+ new ArrayDeque[]{new ArrayDeque(), new ArrayDeque()};
+
+ ResultPartitionWriter partition =
+ new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE));
+ RecordWriter<?> writer = new RecordWriter<>(partition);
+
+ writer.broadcastEvent(EndOfPartitionEvent.INSTANCE);
+
+ // Verify added to all queues
+ assertEquals(1, queues[0].size());
+ assertEquals(1, queues[1].size());
+
+ // these two buffers may share the memory but not the indices!
+ Buffer buffer1 = queues[0].remove();
+ Buffer buffer2 = queues[1].remove();
+ assertEquals(0, buffer1.getReaderIndex());
+ assertEquals(0, buffer2.getReaderIndex());
+ buffer1.setReaderIndex(1);
+ assertEquals("Buffer 2 shares the same reader index as buffer 1", 0, buffer2.getReaderIndex());
+ }
+ /**
+ * Tests that broadcasted records' buffers are independent (in their (reader) indices) once they
+ * are put into the queue for Netty when broadcasting events to multiple channels.
+ */
+ @Test
+ public void testBroadcastEmitBufferIndependence() throws Exception {
+ @SuppressWarnings("unchecked")
+ ArrayDeque<Buffer>[] queues =
+ new ArrayDeque[]{new ArrayDeque(), new ArrayDeque()};
+
+ ResultPartitionWriter partition =
+ new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE));
+ RecordWriter<IntValue> writer = new RecordWriter<>(partition);
+
+ writer.broadcastEmit(new IntValue(0));
+ writer.flush();
+
+ // Verify added to all queues
+ assertEquals(1, queues[0].size());
+ assertEquals(1, queues[1].size());
+
+ // these two buffers may share the memory but not the indices!
+ Buffer buffer1 = queues[0].remove();
+ Buffer buffer2 = queues[1].remove();
+ assertEquals(0, buffer1.getReaderIndex());
+ assertEquals(0, buffer2.getReaderIndex());
+ buffer1.setReaderIndex(1);
+ assertEquals("Buffer 2 shares the same reader index as buffer 1", 0, buffer2.getReaderIndex());
+ }
+
// ---------------------------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
index c190bcf..834ec74 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/ReadOnlySlicedBufferTest.java
@@ -58,7 +58,7 @@ public class ReadOnlySlicedBufferTest {
public void testForwardsIsBuffer() throws IOException {
assertEquals(buffer.isBuffer(), buffer.readOnlySlice().isBuffer());
assertEquals(buffer.isBuffer(), buffer.readOnlySlice(1, 2).isBuffer());
- NetworkBuffer eventBuffer = (NetworkBuffer) EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE);
+ Buffer eventBuffer = EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE);
assertEquals(eventBuffer.isBuffer(), eventBuffer.readOnlySlice().isBuffer());
assertEquals(eventBuffer.isBuffer(), eventBuffer.readOnlySlice(1, 2).isBuffer());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
index 53ecda3..ad0f103 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyMessageSerializationTest.java
@@ -33,16 +33,15 @@ import org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
import org.junit.Test;
-import java.nio.ByteBuffer;
import java.util.Random;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.powermock.api.mockito.PowerMockito.spy;
+/**
+ * Tests for the serialization and deserialization of the various {@link NettyMessage} sub-classes.
+ */
public class NettyMessageSerializationTest {
private final EmbeddedChannel channel = new EmbeddedChannel(
@@ -54,40 +53,8 @@ public class NettyMessageSerializationTest {
@Test
public void testEncodeDecode() {
- {
- Buffer buffer = spy(new NetworkBuffer(MemorySegmentFactory.allocateUnpooledSegment(1024), FreeingBufferRecycler.INSTANCE));
- ByteBuffer nioBuffer = buffer.getNioBuffer(0, buffer.getMaxCapacity());
-
- for (int i = 0; i < 1024; i += 4) {
- nioBuffer.putInt(i);
- }
- buffer.setSize(1024);
-
- NettyMessage.BufferResponse expected = new NettyMessage.BufferResponse(buffer, random.nextInt(), new InputChannelID(), random.nextInt());
- NettyMessage.BufferResponse actual = encodeAndDecode(expected);
-
- // Verify recycle has been called on buffer instance
- verify(buffer, times(1)).recycleBuffer();
-
- final ByteBuf retainedSlice = actual.getNettyBuffer();
-
- // Ensure not recycled and same size as original buffer
- assertEquals(1, retainedSlice.refCnt());
- assertEquals(1024, retainedSlice.readableBytes());
-
- nioBuffer = retainedSlice.nioBuffer();
- for (int i = 0; i < 1024; i += 4) {
- assertEquals(i, nioBuffer.getInt());
- }
-
- // Release the retained slice
- actual.releaseBuffer();
- assertEquals(0, retainedSlice.refCnt());
-
- assertEquals(expected.sequenceNumber, actual.sequenceNumber);
- assertEquals(expected.receiverId, actual.receiverId);
- assertEquals(expected.backlog, actual.backlog);
- }
+ testEncodeDecodeBuffer(false);
+ testEncodeDecodeBuffer(true);
{
{
@@ -170,12 +137,50 @@ public class NettyMessageSerializationTest {
}
}
+ private void testEncodeDecodeBuffer(boolean testReadOnlyBuffer) {
+ NetworkBuffer buffer = new NetworkBuffer(MemorySegmentFactory.allocateUnpooledSegment(1024), FreeingBufferRecycler.INSTANCE);
+
+ for (int i = 0; i < 1024; i += 4) {
+ buffer.writeInt(i);
+ }
+
+ Buffer testBuffer = testReadOnlyBuffer ? buffer.readOnlySlice() : buffer;
+
+ NettyMessage.BufferResponse expected = new NettyMessage.BufferResponse(
+ testBuffer, random.nextInt(), new InputChannelID(), random.nextInt());
+ NettyMessage.BufferResponse actual = encodeAndDecode(expected);
+
+ // Verify recycle has been called on buffer instance (LengthFieldBasedFrameDecoder creates a new one)
+ assertTrue(buffer.isRecycled());
+ assertTrue(testBuffer.isRecycled());
+
+ final ByteBuf retainedSlice = actual.getNettyBuffer();
+
+ // Ensure not recycled and same size as original buffer
+ assertEquals(1, retainedSlice.refCnt());
+ assertEquals(1024, retainedSlice.readableBytes());
+
+ for (int i = 0; i < 1024; i += 4) {
+ assertEquals(i, retainedSlice.readInt());
+ }
+
+ // Release the retained slice
+ actual.releaseBuffer();
+ assertEquals(0, retainedSlice.refCnt());
+ assertTrue(buffer.isRecycled());
+ assertTrue(testBuffer.isRecycled());
+
+ assertEquals(expected.sequenceNumber, actual.sequenceNumber);
+ assertEquals(expected.receiverId, actual.receiverId);
+ assertEquals(expected.backlog, actual.backlog);
+ }
+
@SuppressWarnings("unchecked")
private <T extends NettyMessage> T encodeAndDecode(T msg) {
channel.writeOutbound(msg);
ByteBuf encoded = (ByteBuf) channel.readOutbound();
- channel.writeInbound(encoded);
+ assertTrue(channel.writeInbound(encoded));
return (T) channel.readInbound();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
index 996722f..fff0e22 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferListener;
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.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
import org.apache.flink.runtime.io.network.netty.NettyMessage.AddCredit;
import org.apache.flink.runtime.io.network.netty.NettyMessage.BufferResponse;
@@ -448,7 +449,7 @@ public class PartitionRequestClientHandlerTest {
ByteBuf serialized = resp.write(UnpooledByteBufAllocator.DEFAULT);
// Skip general header bytes
- serialized.readBytes(NettyMessage.HEADER_LENGTH);
+ serialized.readBytes(NettyMessage.FRAME_HEADER_LENGTH);
// Deserialize the bytes again. We have to go this way, because we only partly deserialize
// the header of the response and wait for a buffer from the buffer pool to copy the payload
http://git-wip-us.apache.org/repos/asf/flink/blob/03c797a1/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
index 5c9b5c9..17b87f7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
@@ -22,20 +22,33 @@ import org.apache.flink.runtime.execution.CancelTaskException;
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.ResultPartitionProvider;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartition;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannelID;
+import org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
import org.junit.Test;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+/**
+ * Tests for {@link PartitionRequestQueue}.
+ */
public class PartitionRequestQueueTest {
@Test
@@ -71,4 +84,110 @@ public class PartitionRequestQueueTest {
NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg;
assertTrue(err.cause instanceof CancelTaskException);
}
+
+ /**
+ * Tests {@link PartitionRequestQueue} buffer writing with default buffers.
+ */
+ @Test
+ public void testDefaultBufferWriting() throws Exception {
+ testBufferWriting(new DefaultBufferResultSubpartitionView(2));
+ }
+
+ /**
+ * Tests {@link PartitionRequestQueue} buffer writing with read-only buffers.
+ */
+ @Test
+ public void testReadOnlyBufferWriting() throws Exception {
+ testBufferWriting(new ReadOnlyBufferResultSubpartitionView(2));
+ }
+
+ private void testBufferWriting(ResultSubpartitionView view) throws IOException {
+ // setup
+ ResultPartitionProvider partitionProvider =
+ (partitionId, index, availabilityListener) -> view;
+
+ final InputChannelID receiverId = new InputChannelID();
+ final PartitionRequestQueue queue = new PartitionRequestQueue();
+ final SequenceNumberingViewReader reader = new SequenceNumberingViewReader(receiverId, queue);
+ final EmbeddedChannel channel = new EmbeddedChannel(queue);
+
+ reader.requestSubpartitionView(partitionProvider, new ResultPartitionID(), 0);
+
+ // notify about buffer availability and encode one buffer
+ reader.notifyBuffersAvailable(1);
+
+ channel.runPendingTasks();
+
+ Object read = channel.readOutbound();
+ assertNotNull(read);
+ if (read instanceof NettyMessage.ErrorResponse) {
+ ((NettyMessage.ErrorResponse) read).cause.printStackTrace();
+ }
+ assertThat(read, instanceOf(NettyMessage.BufferResponse.class));
+ read = channel.readOutbound();
+ assertNull(read);
+ }
+
+ private static class DefaultBufferResultSubpartitionView extends NoOpResultSubpartitionView {
+ /** Number of buffer in the backlog to report with every {@link #getNextBuffer()} call. */
+ final int buffersInBacklog;
+
+ private DefaultBufferResultSubpartitionView(int buffersInBacklog) {
+ this.buffersInBacklog = buffersInBacklog;
+ }
+
+ @Nullable
+ @Override
+ public ResultSubpartition.BufferAndBacklog getNextBuffer() {
+ return new ResultSubpartition.BufferAndBacklog(
+ TestBufferFactory.createBuffer(10),
+ buffersInBacklog);
+ }
+ }
+
+ private static class ReadOnlyBufferResultSubpartitionView extends NoOpResultSubpartitionView {
+ /** Number of buffer in the backlog to report with every {@link #getNextBuffer()} call. */
+ final int buffersInBacklog;
+
+ private ReadOnlyBufferResultSubpartitionView(int buffersInBacklog) {
+ this.buffersInBacklog = buffersInBacklog;
+ }
+
+ @Nullable
+ @Override
+ public ResultSubpartition.BufferAndBacklog getNextBuffer() {
+ return new ResultSubpartition.BufferAndBacklog(
+ TestBufferFactory.createBuffer(10).readOnlySlice(),
+ buffersInBacklog);
+ }
+ }
+
+ private static class NoOpResultSubpartitionView implements ResultSubpartitionView {
+ @Nullable
+ public ResultSubpartition.BufferAndBacklog getNextBuffer() {
+ return null;
+ }
+
+ @Override
+ public void notifyBuffersAvailable(long buffers) {
+ }
+
+ @Override
+ public void releaseAllResources() {
+ }
+
+ @Override
+ public void notifySubpartitionConsumed() {
+ }
+
+ @Override
+ public boolean isReleased() {
+ return true;
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ return null;
+ }
+ }
}