PartitionRequestQueue extends ChannelInboundHandlerAdapter { public void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId) throws Exception { ctx.pipeline().fireUserEventTriggered(new SequenceNumberingSubpartitionView(partitionQueue, receiverId)); }  @Override void channelRegistered(final ChannelHandlerContext ctx); void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId); 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(); EmbeddedChannel ch = new EmbeddedChannel(queue); ResultSubpartitionView view = mock(ResultSubpartitionView.class); when(view.isReleased()).thenReturn(true); when(view.getFailureCause()).thenReturn(new RuntimeException("Expected test exception")); queue.enqueue(view, new InputChannelID()); ch.runPendingTasks(); Object msg = ch.readOutbound(); assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class); NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg; assertTrue(err.cause instanceof CancelTaskException); }