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 ] ; } }
Requests a remote intermediate result partition queue .