@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);
	// Enqueue the erroneous view
	queue.notifyReaderNonEmpty(seqView);
	ch.runPendingTasks();
	// Read the enqueued msg
	Object msg = ch.readOutbound();
	assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class);
	NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg;
	assertTrue(err.cause instanceof CancelTaskException);
}