Fix leak and corruption bugs in CompositeByteBuf (netty#8438)

Motivation:

I came across two bugs:
- Components removed due to capacity reduction aren't released
- Offsets aren't set correctly on empty components that are added
between existing components

Modifications:

Add unit tests which expose these bugs, fix them.

Result:

Bugs are fixed

buffer/src/main/java/io/netty/buffer/CompositeByteBuf.java

@@ -260,24 +260,18 @@ private int addComponent0(boolean increaseWriterIndex, int cIndex, ByteBuf buffe
             // No need to consolidate - just add a component to the list.
             @SuppressWarnings("deprecation")
             Component c = new Component(buffer.order(ByteOrder.BIG_ENDIAN).slice());
-            if (cIndex == components.size()) {
-                wasAdded = components.add(c);
-                if (cIndex == 0) {
-                    c.endOffset = readableBytes;
-                } else {
-                    Component prev = components.get(cIndex - 1);
-                    c.offset = prev.endOffset;
-                    c.endOffset = c.offset + readableBytes;
-                }
+            components.add(cIndex, c);
+            wasAdded = true;
+            if (readableBytes > 0 && cIndex < components.size() - 1) {
+                updateComponentOffsets(cIndex);
+            } else if (cIndex == 0) {
+                c.endOffset = readableBytes;
             } else {
-                components.add(cIndex, c);
-                wasAdded = true;
-                if (readableBytes != 0) {
-                    updateComponentOffsets(cIndex);
-                }
+                c.offset = components.get(cIndex - 1).endOffset;
+                c.endOffset = c.offset + readableBytes;
             }
             if (increaseWriterIndex) {
-                writerIndex(writerIndex() + buffer.readableBytes());
+                writerIndex(writerIndex() + readableBytes);
             }
             return cIndex;
         } finally {
@@ -663,6 +657,7 @@ public CompositeByteBuf capacity(int newCapacity) {
                 Component c = i.previous();
                 if (bytesToTrim >= c.length) {
                     bytesToTrim -= c.length;
+                    c.freeIfNecessary();
                     i.remove();
                     continue;
                 }
@@ -1635,6 +1630,7 @@ public CompositeByteBuf discardReadBytes() {
         if (adjustment == c.length) {
             // new slice would be empty, so remove instead
             firstComponentId++;
+            c.freeIfNecessary();
         } else {
             Component newC = new Component(c.buf.slice(adjustment, c.length - adjustment));
             components.set(firstComponentId, newC);

buffer/src/test/java/io/netty/buffer/AbstractCompositeByteBufTest.java

@@ -1018,6 +1018,33 @@ public void testAddEmptyBufferInMiddle() {
         cbuf.release();
     }
 
+    @Test
+    public void testInsertEmptyBufferInMiddle() {
+        CompositeByteBuf cbuf = compositeBuffer();
+        ByteBuf buf1 = buffer().writeByte((byte) 1);
+        cbuf.addComponent(true, buf1);
+        ByteBuf buf2 = buffer().writeByte((byte) 2);
+        cbuf.addComponent(true, buf2);
+
+        // insert empty one between the first two
+        cbuf.addComponent(true, 1, EMPTY_BUFFER);
+
+        assertEquals(2, cbuf.readableBytes());
+        assertEquals((byte) 1, cbuf.readByte());
+        assertEquals((byte) 2, cbuf.readByte());
+
+        assertEquals(2, cbuf.capacity());
+        assertEquals(3, cbuf.numComponents());
+
+        byte[] dest = new byte[2];
+        // should skip over the empty one, not throw a java.lang.Error :)
+        cbuf.getBytes(0, dest);
+
+        assertArrayEquals(new byte[] {1, 2}, dest);
+
+        cbuf.release();
+    }
+
     @Test
     public void testIterator() {
         CompositeByteBuf cbuf = compositeBuffer();
@@ -1117,6 +1144,29 @@ public void testReleasesItsComponents() {
         assertEquals(0, buffer.refCnt());
     }
 
+    @Test
+    public void testReleasesOnShrink() {
+
+        ByteBuf b1 = Unpooled.buffer(2).writeShort(1);
+        ByteBuf b2 = Unpooled.buffer(2).writeShort(2);
+
+        // composite takes ownership of s1 and s2
+        ByteBuf composite = Unpooled.compositeBuffer()
+            .addComponents(b1, b2);
+
+        assertEquals(4, composite.capacity());
+
+        // reduce capacity down to two, will drop the second component
+        composite.capacity(2);
+        assertEquals(2, composite.capacity());
+
+        // releasing composite should release the components
+        composite.release();
+        assertEquals(0, composite.refCnt());
+        assertEquals(0, b1.refCnt());
+        assertEquals(0, b2.refCnt());
+    }
+
     @Test
     public void testAllocatorIsSameWhenCopy() {
         testAllocatorIsSameWhenCopy(false);