zhijiangW commented on a change in pull request #7368: [FLINK-10742][network]
Let Netty use Flink's buffers directly in credit-based mode
URL: https://github.com/apache/flink/pull/7368#discussion_r388168150
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/CreditBasedPartitionRequestClientHandlerTest.java
##########
@@ -419,6 +459,99 @@ public void testNotifyCreditAvailableAfterReleased()
throws Exception {
}
}
+ @Test
+ public void testChannelReleasedBeforeDecodingBufferResponse() throws
Exception {
+ testChannelReleasedOrRemovedBeforeDecodingBufferResponse(false);
+ }
+
+ @Test
+ public void testChannelRemovedBeforeDecodingBufferResponse() throws
Exception {
+ testChannelReleasedOrRemovedBeforeDecodingBufferResponse(true);
+ }
+
+ @Test
+ public void testChannelReleasedBeforeReceivingBufferResponse() throws
Exception {
+
testChannelReleasedOrRemovedBeforeReceivingBufferResponse(false);
+ }
+
+ @Test
+ public void testChannelRemovedBeforeReceivingBufferResponse() throws
Exception {
+ testChannelReleasedOrRemovedBeforeReceivingBufferResponse(true);
+ }
+
+ private void
testChannelReleasedOrRemovedBeforeDecodingBufferResponse(boolean isRemoved)
throws Exception {
+ int bufferSize = 1024;
+
+ NetworkBufferPool networkBufferPool = new NetworkBufferPool(10,
bufferSize, 2);
+ SingleInputGate inputGate = createSingleInputGate(1);
+ RemoteInputChannel inputChannel = new InputChannelBuilder()
+ .setMemorySegmentProvider(networkBufferPool)
+ .buildRemoteAndSetToGate(inputGate);
+ inputGate.assignExclusiveSegments();
+
+ CreditBasedPartitionRequestClientHandler handler = new
CreditBasedPartitionRequestClientHandler();
+ handler.addInputChannel(inputChannel);
+
+ try {
+ BufferResponse bufferResponse = createBufferResponse(
+ TestBufferFactory.createBuffer(bufferSize),
+ 0,
+ inputChannel.getInputChannelId(),
+ 1,
+ new NetworkBufferAllocator(handler));
+
+ // Release the channel.
+ inputGate.close();
+ if (isRemoved) {
+ handler.removeInputChannel(inputChannel);
+ }
+
+ handler.channelRead(null, bufferResponse);
+
+ assertEquals(0,
inputChannel.getNumberOfQueuedBuffers());
+ assertNotNull(bufferResponse.getBuffer());
+ assertTrue(bufferResponse.getBuffer().isRecycled());
+ } finally {
+ releaseResource(inputGate, networkBufferPool);
+ }
+ }
+
+ private void
testChannelReleasedOrRemovedBeforeReceivingBufferResponse(boolean isRemoved)
throws Exception {
+ int bufferSize = 1024;
+
+ NetworkBufferPool networkBufferPool = new NetworkBufferPool(10,
bufferSize, 2);
+ SingleInputGate inputGate = createSingleInputGate(1);
+ RemoteInputChannel inputChannel = new InputChannelBuilder()
+ .setMemorySegmentProvider(networkBufferPool)
+ .buildRemoteAndSetToGate(inputGate);
+ inputGate.assignExclusiveSegments();
+
+ CreditBasedPartitionRequestClientHandler handler = spy(new
CreditBasedPartitionRequestClientHandler());
+ handler.addInputChannel(inputChannel);
+
+ try {
+ // Release the channel.
+ inputGate.close();
+ if (isRemoved) {
+ handler.removeInputChannel(inputChannel);
+ }
+
+ BufferResponse bufferResponse = createBufferResponse(
+ TestBufferFactory.createBuffer(bufferSize),
+ 0,
+ inputChannel.getInputChannelId(),
+ 1,
+ new NetworkBufferAllocator(handler));
+ handler.channelRead(null, bufferResponse);
+
+ assertEquals(0,
inputChannel.getNumberOfQueuedBuffers());
+ assertNull(bufferResponse.getBuffer());
+ verify(handler,
times(1)).cancelRequestFor(eq(inputChannel.getInputChannelId()));
Review comment:
The above testChannelReleasedOrRemovedBeforeDecodingBufferResponse should
also cover this verify. Also it is not very suggested using `spy` to verify it.
We can also verify either via exposing `cancelled` map from
CreditBasedPartitionRequestClientHandler or reading the `CancelPartition`
message from the `EmbeddedChannel`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
