PartitionRequestQueue extends ChannelInboundHandlerAdapter { void notifyReaderNonEmpty(final SequenceNumberingViewReader reader) { ctx.executor().execute(new Runnable() { @Override public void run() { ctx.pipeline().fireUserEventTriggered(reader); } }); }  @Override void channelRegistered(final ChannelHandlerContext ctx); void cancel(InputChannelID receiverId); void close(); @Override void userEventTriggered(ChannelHandlerContext ctx, Object msg); @Override void channelWritabilityChanged(ChannelHandlerContext ctx); @Override void channelInactive(ChannelHandlerContext ctx); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);  }
@Test public void testProducerFailedException() throws Exception { PartitionRequestQueue queue = new PartitionRequestQueue(); ResultPartitionProvider partitionProvider = mock(ResultPartitionProvider.class); ResultPartitionID rpid = new ResultPartitionID(); ResultSubpartitionView view = mock(ResultSubpartitionView.class); when(view.isReleased()).thenReturn(true); when(view.getFailureCause()).thenReturn(new RuntimeException("Expected test exception")); when(partitionProvider.createSubpartitionView( eq(rpid), eq(0), any(BufferAvailabilityListener.class))).thenReturn(view); EmbeddedChannel ch = new EmbeddedChannel(queue); SequenceNumberingViewReader seqView = new SequenceNumberingViewReader(new InputChannelID(), queue); seqView.requestSubpartitionView(partitionProvider, rpid, 0); queue.notifyReaderNonEmpty(seqView); ch.runPendingTasks(); Object msg = ch.readOutbound(); assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class); NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg; assertTrue(err.cause instanceof CancelTaskException); }