@Test
@SuppressWarnings("unchecked")
public void testAutoReadAfterUnsuccessfulStagedMessage() throws Exception {

	PartitionRequestClientHandler handler = new PartitionRequestClientHandler();
	EmbeddedChannel channel = new EmbeddedChannel(handler);
	final AtomicReference<BufferListener> listener = new AtomicReference<>();
	BufferProvider bufferProvider = mock(BufferProvider.class);
	when(bufferProvider.addBufferListener(any(BufferListener.class))).thenAnswer(new Answer<Boolean>() {
		@Override
		@SuppressWarnings("unchecked")
		public Boolean answer(InvocationOnMock invocation) throws Throwable {
			listener.set((BufferListener) invocation.getArguments()[0]);
			return true;
		}
	});
	when(bufferProvider.requestBuffer()).thenReturn(null);
	InputChannelID channelId = new InputChannelID(0, 0);
	RemoteInputChannel inputChannel = mock(RemoteInputChannel.class);
	when(inputChannel.getInputChannelId()).thenReturn(channelId);
	// The 3rd staged msg has a null buffer provider
	when(inputChannel.getBufferProvider()).thenReturn(bufferProvider, bufferProvider, null);
	handler.addInputChannel(inputChannel);
	BufferResponse msg = createBufferResponse(createBuffer(true), 0, channelId);
	// Write 1st buffer msg. No buffer is available, therefore the buffer
	// should be staged and auto read should be set to false.
	assertTrue(channel.config().isAutoRead());
	channel.writeInbound(msg);
	// No buffer available, auto read false
	assertFalse(channel.config().isAutoRead());
	// Write more buffers... all staged.
	msg = createBufferResponse(createBuffer(true), 1, channelId);
	channel.writeInbound(msg);
	msg = createBufferResponse(createBuffer(true), 2, channelId);
	channel.writeInbound(msg);
	// Notify about buffer => handle 1st msg
	Buffer availableBuffer = createBuffer(false);
	listener.get().notifyBufferAvailable(availableBuffer);
	// Start processing of staged buffers (in run pending tasks). Make
	// sure that the buffer provider acts like it's destroyed.
	when(bufferProvider.addBufferListener(any(BufferListener.class))).thenReturn(false);
	when(bufferProvider.isDestroyed()).thenReturn(true);
	// Execute all tasks that are scheduled in the event loop. Further
	// eventLoop().execute() calls are directly executed, if they are
	// called in the scope of this call.
	channel.runPendingTasks();
	assertTrue(channel.config().isAutoRead());
}