@Test public void testDoublePartitionRequest() throws Exception { final PartitionRequestClientHandler handler = new PartitionRequestClientHandler(); final EmbeddedChannel channel = new EmbeddedChannel(handler); final PartitionRequestClient client = new PartitionRequestClient( channel, handler, mock(ConnectionID.class), mock(PartitionRequestClientFactory.class)); final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32); final SingleInputGate inputGate = createSingleInputGate(); final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client); try { final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6); inputGate.setBufferPool(bufferPool); final int numExclusiveBuffers = 2; inputGate.assignExclusiveSegments(networkBufferPool, numExclusiveBuffers); inputChannel.requestSubpartition(0); assertTrue(channel.isWritable()); Object readFromOutbound = channel.readOutbound(); assertThat(readFromOutbound, instanceOf(PartitionRequest.class)); assertEquals(inputChannel.getInputChannelId(), ((PartitionRequest) readFromOutbound).receiverId); assertEquals(numExclusiveBuffers, ((PartitionRequest) readFromOutbound).credit); assertNull(channel.readOutbound()); } finally { inputGate.releaseAllResources(); networkBufferPool.destroyAllBufferPools(); networkBufferPool.destroy(); } }
public ChannelFuture requestSubpartition( final ResultPartitionID partitionId, final int subpartitionIndex, final RemoteInputChannel inputChannel, int delayMs) throws IOException { checkNotClosed(); LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.", subpartitionIndex, partitionId, delayMs); clientHandler.addInputChannel(inputChannel); final PartitionRequest request = new PartitionRequest( partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit()); final ChannelFutureListener listener = new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { clientHandler.removeInputChannel(inputChannel); SocketAddress remoteAddr = future.channel().remoteAddress(); inputChannel.onError( new LocalTransportException( String.format("Sending the partition request to '%s' failed.", remoteAddr), future.channel().localAddress(), future.cause() )); } } }; if (delayMs == 0) { ChannelFuture f = tcpChannel.writeAndFlush(request); f.addListener(listener); return f; } else { final ChannelFuture[] f = new ChannelFuture[1]; tcpChannel.eventLoop().schedule(new Runnable() { @Override public void run() { f[0] = tcpChannel.writeAndFlush(request); f[0].addListener(listener); } }, delayMs, TimeUnit.MILLISECONDS); return f[0]; } }
PartitionRequestClient { public ChannelFuture requestSubpartition( final ResultPartitionID partitionId, final int subpartitionIndex, final RemoteInputChannel inputChannel, int delayMs) throws IOException { checkNotClosed(); LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.", subpartitionIndex, partitionId, delayMs); clientHandler.addInputChannel(inputChannel); final PartitionRequest request = new PartitionRequest( partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit()); final ChannelFutureListener listener = new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { clientHandler.removeInputChannel(inputChannel); SocketAddress remoteAddr = future.channel().remoteAddress(); inputChannel.onError( new LocalTransportException( String.format("Sending the partition request to '%s' failed.", remoteAddr), future.channel().localAddress(), future.cause() )); } } }; if (delayMs == 0) { ChannelFuture f = tcpChannel.writeAndFlush(request); f.addListener(listener); return f; } else { final ChannelFuture[] f = new ChannelFuture[1]; tcpChannel.eventLoop().schedule(new Runnable() { @Override public void run() { f[0] = tcpChannel.writeAndFlush(request); f[0].addListener(listener); } }, delayMs, TimeUnit.MILLISECONDS); return f[0]; } } }
PartitionRequestClient { public ChannelFuture requestSubpartition( final ResultPartitionID partitionId, final int subpartitionIndex, final RemoteInputChannel inputChannel, int delayMs) throws IOException { checkNotClosed(); LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.", subpartitionIndex, partitionId, delayMs); clientHandler.addInputChannel(inputChannel); final PartitionRequest request = new PartitionRequest( partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit()); final ChannelFutureListener listener = new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { clientHandler.removeInputChannel(inputChannel); SocketAddress remoteAddr = future.channel().remoteAddress(); inputChannel.onError( new LocalTransportException( String.format("Sending the partition request to '%s' failed.", remoteAddr), future.channel().localAddress(), future.cause() )); } } }; if (delayMs == 0) { ChannelFuture f = tcpChannel.writeAndFlush(request); f.addListener(listener); return f; } else { final ChannelFuture[] f = new ChannelFuture[1]; tcpChannel.eventLoop().schedule(new Runnable() { @Override public void run() { f[0] = tcpChannel.writeAndFlush(request); f[0].addListener(listener); } }, delayMs, TimeUnit.MILLISECONDS); return f[0]; } } PartitionRequestClient(
			Channel tcpChannel,
			NetworkClientHandler clientHandler,
			ConnectionID connectionId,
			PartitionRequestClientFactory clientFactory); }
PartitionRequestClient { public ChannelFuture requestSubpartition( final ResultPartitionID partitionId, final int subpartitionIndex, final RemoteInputChannel inputChannel, int delayMs) throws IOException { checkNotClosed(); LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.", subpartitionIndex, partitionId, delayMs); clientHandler.addInputChannel(inputChannel); final PartitionRequest request = new PartitionRequest( partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit()); final ChannelFutureListener listener = new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { clientHandler.removeInputChannel(inputChannel); SocketAddress remoteAddr = future.channel().remoteAddress(); inputChannel.onError( new LocalTransportException( String.format("Sending the partition request to '%s' failed.", remoteAddr), future.channel().localAddress(), future.cause() )); } } }; if (delayMs == 0) { ChannelFuture f = tcpChannel.writeAndFlush(request); f.addListener(listener); return f; } else { final ChannelFuture[] f = new ChannelFuture[1]; tcpChannel.eventLoop().schedule(new Runnable() { @Override public void run() { f[0] = tcpChannel.writeAndFlush(request); f[0].addListener(listener); } }, delayMs, TimeUnit.MILLISECONDS); return f[0]; } } PartitionRequestClient(
			Channel tcpChannel,
			NetworkClientHandler clientHandler,
			ConnectionID connectionId,
			PartitionRequestClientFactory clientFactory); ChannelFuture requestSubpartition(
			final ResultPartitionID partitionId,
			final int subpartitionIndex,
			final RemoteInputChannel inputChannel,
			int delayMs); void sendTaskEvent(ResultPartitionID partitionId, TaskEvent event, final RemoteInputChannel inputChannel); void notifyCreditAvailable(RemoteInputChannel inputChannel); void close(RemoteInputChannel inputChannel); }
PartitionRequestClient { public ChannelFuture requestSubpartition( final ResultPartitionID partitionId, final int subpartitionIndex, final RemoteInputChannel inputChannel, int delayMs) throws IOException { checkNotClosed(); LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.", subpartitionIndex, partitionId, delayMs); clientHandler.addInputChannel(inputChannel); final PartitionRequest request = new PartitionRequest( partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit()); final ChannelFutureListener listener = new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { clientHandler.removeInputChannel(inputChannel); SocketAddress remoteAddr = future.channel().remoteAddress(); inputChannel.onError( new LocalTransportException( String.format("Sending the partition request to '%s' failed.", remoteAddr), future.channel().localAddress(), future.cause() )); } } }; if (delayMs == 0) { ChannelFuture f = tcpChannel.writeAndFlush(request); f.addListener(listener); return f; } else { final ChannelFuture[] f = new ChannelFuture[1]; tcpChannel.eventLoop().schedule(new Runnable() { @Override public void run() { f[0] = tcpChannel.writeAndFlush(request); f[0].addListener(listener); } }, delayMs, TimeUnit.MILLISECONDS); return f[0]; } } PartitionRequestClient(
			Channel tcpChannel,
			NetworkClientHandler clientHandler,
			ConnectionID connectionId,
			PartitionRequestClientFactory clientFactory); ChannelFuture requestSubpartition(
			final ResultPartitionID partitionId,
			final int subpartitionIndex,
			final RemoteInputChannel inputChannel,
			int delayMs); void sendTaskEvent(ResultPartitionID partitionId, TaskEvent event, final RemoteInputChannel inputChannel); void notifyCreditAvailable(RemoteInputChannel inputChannel); void close(RemoteInputChannel inputChannel);  }