@Test
public void errorDuringDeliveryShouldReturnCorrectNumberOfBytes() throws Exception {

    final ByteBuf data = dummyData();
    final int padding = 10;
    final AtomicInteger unprocessed = new AtomicInteger(data.readableBytes() + padding);
    doAnswer(new Answer<Integer>() {
        @Override
        public Integer answer(InvocationOnMock in) throws Throwable {
            return unprocessed.get();
        }
    }).when(localFlow).unconsumedBytes(eq(stream));
    doAnswer(new Answer<Void>() {
        @Override
        public Void answer(InvocationOnMock in) throws Throwable {
            int delta = (Integer) in.getArguments()[1];
            int newValue = unprocessed.addAndGet(-delta);
            if (newValue < 0) {
                throw new RuntimeException("Returned too many bytes");
            }
            return null;
        }
    }).when(localFlow).consumeBytes(eq(stream), anyInt());
    // When the listener callback is called, process a few bytes and then throw.
    doAnswer(new Answer<Integer>() {
        @Override
        public Integer answer(InvocationOnMock in) throws Throwable {
            localFlow.consumeBytes(stream, 4);
            throw new RuntimeException("Fake Exception");
        }
    }).when(listener).onDataRead(eq(ctx), eq(STREAM_ID), any(ByteBuf.class), eq(10), eq(true));
    try {
        decode().onDataRead(ctx, STREAM_ID, data, padding, true);
        fail("Expected exception");
    } catch (RuntimeException cause) {
        verify(localFlow)
                .receiveFlowControlledFrame(eq(stream), eq(data), eq(padding), eq(true));
        verify(lifecycleManager).closeStreamRemote(eq(stream), eq(future));
        verify(listener).onDataRead(eq(ctx), eq(STREAM_ID), eq(data), eq(padding), eq(true));
        assertEquals(0, localFlow.unconsumedBytes(stream));
    } finally {
        data.release();
    }
}