@Test public void testNotifyCreditAvailableAfterReleased() throws Exception { final CreditBasedPartitionRequestClientHandler handler = new CreditBasedPartitionRequestClientHandler(); final EmbeddedChannel channel = new EmbeddedChannel(handler); final PartitionRequestClient client = new NettyPartitionRequestClient( channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class)); final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32, 2); final SingleInputGate inputGate = createSingleInputGate(1); final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client, networkBufferPool); try { final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6); inputGate.setBufferPool(bufferPool); inputGate.assignExclusiveSegments(); inputChannel.requestSubpartition(0); Object readFromOutbound = channel.readOutbound(); assertThat(readFromOutbound, instanceOf(PartitionRequest.class)); assertEquals(2, ((PartitionRequest) readFromOutbound).credit); final BufferResponse bufferResponse = createBufferResponse( TestBufferFactory.createBuffer(32), 0, inputChannel.getInputChannelId(), 1); handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse); assertEquals(2, inputChannel.getUnannouncedCredit()); inputGate.close(); readFromOutbound = channel.readOutbound(); assertThat(readFromOutbound, instanceOf(CloseRequest.class)); channel.runPendingTasks(); assertNull(channel.readOutbound()); } finally { inputGate.close(); networkBufferPool.destroyAllBufferPools(); networkBufferPool.destroy(); } }
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { decodeMsg(msg); } catch (Throwable t) { notifyAllChannelsOfErrorAndClose(t); } }
CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { decodeMsg(msg); } catch (Throwable t) { notifyAllChannelsOfErrorAndClose(t); } } }
CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { decodeMsg(msg); } catch (Throwable t) { notifyAllChannelsOfErrorAndClose(t); } }  }
CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { decodeMsg(msg); } catch (Throwable t) { notifyAllChannelsOfErrorAndClose(t); } }  @Override void addInputChannel(RemoteInputChannel listener); @Override void removeInputChannel(RemoteInputChannel listener); @Override void cancelRequestFor(InputChannelID inputChannelId); @Override void notifyCreditAvailable(final RemoteInputChannel inputChannel); @Override void channelActive(final ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void userEventTriggered(ChannelHandlerContext ctx, Object msg); @Override void channelWritabilityChanged(ChannelHandlerContext ctx); }
CreditBasedPartitionRequestClientHandler extends ChannelInboundHandlerAdapter implements NetworkClientHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { decodeMsg(msg); } catch (Throwable t) { notifyAllChannelsOfErrorAndClose(t); } }  @Override void addInputChannel(RemoteInputChannel listener); @Override void removeInputChannel(RemoteInputChannel listener); @Override void cancelRequestFor(InputChannelID inputChannelId); @Override void notifyCreditAvailable(final RemoteInputChannel inputChannel); @Override void channelActive(final ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void userEventTriggered(ChannelHandlerContext ctx, Object msg); @Override void channelWritabilityChanged(ChannelHandlerContext ctx);  }