Repository: netty/netty
Issue: remove handler cause ByteToMessageDecoder out disorder

### remove handler cause ByteToMessageDecoder out disorder
	
	I input byte stream [1,2,3,4,5], when i read `4` then remove  ByteToMessageDecoder,
	 the out is stream [1,2,3,5,4]
	
	Here are the test cases，the cases will fail


```
 @Test
    public void testDisorder(){
        ByteToMessageDecoder decoder = new ByteToMessageDecoder() {
            int count = 0;

            //read 4 byte then remove this decoder
            @Override
            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                out.add(in.readBytes(1));
                if (++count >= 4) {
                    ctx.pipeline().remove(this);
                }
            }
        };

        EmbeddedChannel channel = new EmbeddedChannel(decoder);
        assertTrue(channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1, 2, 3, 4, 5})));
        assertEquals(1, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(2, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(3, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(4, ((ByteBuf) channel.readInbound()).readByte());
        assertEquals(5, ((ByteBuf) channel.readInbound()).readByte());
        assertFalse(channel.finish());
    }


```
### netty 4.1

### java8

### window10

Issue comments:
@normanmaurer thanks
Hi, @switchYello. I think this is expected behavior, see 
 https://github.com/netty/netty/blob/2e5dd288008d4e674f53beaf8d323595813062fb/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L432-L436
@amizurov  

	I see that code annotation and iss,but i think, 
	code 'if (ctx.isRemoved()) {break;}' is to exit the 'while (in.isReadable())' loop,it work good.

	my ISS means,when we remove this decoder, 'webInStream' has byte data and 'out List' has decoded data,we should fire 'out List' before 'webInStream'
	Otherwise, it will lead to disorder.

```
 while (in.isReadable()) {
                int outSize = out.size();

                if (outSize > 0) {
                    fireChannelRead(ctx, out, outSize);
                    out.clear();

                    // Check if this handler was removed before continuing with decoding.
                    // If it was removed, it is not safe to continue to operate on the buffer.
                    //
                    // See:
                    // - https://github.com/netty/netty/issues/4635
                    if (ctx.isRemoved()) {
                        break;
                    }
                    outSize = 0;
                }

```


@switchYello ohh missed, that is output list does not drain after handler removed.

Pull request review comments:
nit: add whitespace after each `,`
You will need to call `release()` on each of the `ByteBuf` instances... Alternative you could change your decoder to just do:

```
out.add(in.readByte());
```

And then assert the returned `byte`.
nit: can we also add `assertFalse(buffer5.isReadable());` here to ensure we not produce more then expected ?
@switchYello let me know once you did address this and I will merge. 

Patch that resolved the issue:
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
index 96cd7b3b99b..361863cbafa 100644
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -81,7 +81,7 @@ public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in)
             try {
                 final ByteBuf buffer;
                 if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
-                    || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
+                        || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
                     // Expand cumulation (by replace it) when either there is not more room in the buffer
                     // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
                     // duplicate().retain() or if its read-only.
@@ -507,6 +507,8 @@ final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in,
             boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
             decodeState = STATE_INIT;
             if (removePending) {
+                fireChannelRead(ctx, out, out.size());
+                out.clear();
                 handlerRemoved(ctx);
             }
         }


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
index 875de766625..cd9ee370831 100644
--- a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
@@ -399,4 +399,31 @@ public void read(ChannelHandlerContext ctx) throws Exception {
         }
         assertFalse(channel.finish());
     }
+
+    @Test
+    public void testDisorder() {
+        ByteToMessageDecoder decoder = new ByteToMessageDecoder() {
+            int count;
+
+            //read 4 byte then remove this decoder
+            @Override
+            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+                out.add(in.readByte());
+                if (++count >= 4) {
+                    ctx.pipeline().remove(this);
+                }
+            }
+        };
+        EmbeddedChannel channel = new EmbeddedChannel(decoder);
+        assertTrue(channel.writeInbound(Unpooled.wrappedBuffer(new byte[]{1, 2, 3, 4, 5})));
+        assertEquals((byte) 1,  channel.readInbound());
+        assertEquals((byte) 2,  channel.readInbound());
+        assertEquals((byte) 3,  channel.readInbound());
+        assertEquals((byte) 4,  channel.readInbound());
+        ByteBuf buffer5 = channel.readInbound();
+        assertEquals((byte) 5, buffer5.readByte());
+        assertFalse(buffer5.isReadable());
+        assertTrue(buffer5.release());
+        assertFalse(channel.finish());
+    }
 }