Break up the monolithic BufferTest into multiple test classes

Then run those tests in independent surefire forks.
This should allow Maven to hold on to less test metadata, and cope better with the large number of tests.

Dockerfile

@@ -34,6 +34,6 @@ RUN mvn dependency:go-offline surefire:test checkstyle:check -ntp
 # Copy over the project code and run our build
 COPY . .
 # Make sure Maven has enough memory to keep track of all the tests we run
-ENV MAVEN_OPTS="-Xmx3g -XX:+HeapDumpOnOutOfMemoryError"
+ENV MAVEN_OPTS="-Xmx4g -XX:+HeapDumpOnOutOfMemoryError"
 # Run tests
 CMD mvn verify -o -B -C -T1C -fae -nsu -npu

pom.xml

@@ -75,15 +75,12 @@
     <surefire.version>3.0.0-M5</surefire.version>
     <skipTests>false</skipTests>
     <argLine.common>
-      -server
-      -dsa -da -ea:io.netty...
       -XX:+HeapDumpOnOutOfMemoryError
-      -Xmx3g
+      -Xmx2g
       -Dio.netty.tryReflectionSetAccessible=true
       --add-opens java.base/java.nio=io.netty.common
+      --add-modules jdk.incubator.foreign
     </argLine.common>
-<!--    <argLine.printGC>-XX:+PrintGCDetails</argLine.printGC>-->
-    <argLine.printGC/>
   </properties>
 
   <build>
@@ -173,9 +170,12 @@
           <includes>
             <include>**/*Test*.java</include>
           </includes>
-          <argLine>${argLine.common} ${argLine.printGC} --add-modules jdk.incubator.foreign</argLine>
+          <argLine>${argLine.common}</argLine>
           <!-- Ensure the whole stacktrace is preserved when an exception is thrown. See https://issues.apache.org/jira/browse/SUREFIRE-1457 -->
           <trimStackTrace>false</trimStackTrace>
+          <forkedProcessExitTimeoutInSeconds>600</forkedProcessExitTimeoutInSeconds>
+          <!-- Not reusing forks lets us be more frugal with memory. We have a lot of tests, and they need a lot of metadata -->
+          <reuseForks>false</reuseForks>
           <systemProperties>
             <sample>nosample</sample>
           </systemProperties>

src/test/java/io/netty/buffer/api/BufferBulkAccessTest.java

@@ -0,0 +1,304 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteBuffer;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class BufferBulkAccessTest extends BufferTestSupport {
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void fill(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16)) {
+            assertThat(buf.fill((byte) 0xA5)).isSameAs(buf);
+            buf.writerOffset(16);
+            assertEquals(0xA5A5A5A5_A5A5A5A5L, buf.readLong());
+            assertEquals(0xA5A5A5A5_A5A5A5A5L, buf.readLong());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoByteArray(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.order(BIG_ENDIAN).writeLong(0x0102030405060708L);
+            byte[] array = new byte[8];
+            buf.copyInto(0, array, 0, array.length);
+            assertThat(array).containsExactly(0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08);
+
+            buf.writerOffset(0).order(LITTLE_ENDIAN).writeLong(0x0102030405060708L);
+            buf.copyInto(0, array, 0, array.length);
+            assertThat(array).containsExactly(0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01);
+
+            array = new byte[6];
+            buf.copyInto(1, array, 1, 3);
+            assertThat(array).containsExactly(0x00, 0x07, 0x06, 0x05, 0x00, 0x00);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoHeapByteBuffer(Fixture fixture) {
+        testCopyIntoByteBuffer(fixture, ByteBuffer::allocate);
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoDirectByteBuffer(Fixture fixture) {
+        testCopyIntoByteBuffer(fixture, ByteBuffer::allocateDirect);
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoOnHeapBuf(Fixture fixture) {
+        testCopyIntoBuf(fixture, BufferAllocator.heap()::allocate);
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoOffHeapBuf(Fixture fixture) {
+        testCopyIntoBuf(fixture, BufferAllocator.direct()::allocate);
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoOnHeapBufSlice(Fixture fixture) {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Scope scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                return scope.add(allocator.allocate(size)).writerOffset(size).slice();
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoOffHeapBufSlice(Fixture fixture) {
+        try (BufferAllocator allocator = BufferAllocator.direct();
+             Scope scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                return scope.add(allocator.allocate(size)).writerOffset(size).slice();
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOnHeapOnHeapBuf(Fixture fixture) {
+        try (var a = BufferAllocator.heap();
+             var b = BufferAllocator.heap()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return Buffer.compose(a, bufFirst, bufSecond);
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOnHeapOffHeapBuf(Fixture fixture) {
+        try (var a = BufferAllocator.heap();
+             var b = BufferAllocator.direct()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return Buffer.compose(a, bufFirst, bufSecond);
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOffHeapOnHeapBuf(Fixture fixture) {
+        try (var a = BufferAllocator.direct();
+             var b = BufferAllocator.heap()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return Buffer.compose(a, bufFirst, bufSecond);
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOffHeapOffHeapBuf(Fixture fixture) {
+        try (var a = BufferAllocator.direct();
+             var b = BufferAllocator.direct()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return Buffer.compose(a, bufFirst, bufSecond);
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOnHeapOnHeapBufSlice(Fixture fixture) {
+        try (var a = BufferAllocator.heap();
+             var b = BufferAllocator.heap();
+             var scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return scope.add(Buffer.compose(a, bufFirst, bufSecond)).writerOffset(size).slice();
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOnHeapOffHeapBufSlice(Fixture fixture) {
+        try (var a = BufferAllocator.heap();
+             var b = BufferAllocator.direct();
+             var scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return scope.add(Buffer.compose(a, bufFirst, bufSecond)).writerOffset(size).slice();
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOffHeapOnHeapBufSlice(Fixture fixture) {
+        try (var a = BufferAllocator.direct();
+             var b = BufferAllocator.heap();
+             var scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return scope.add(Buffer.compose(a, bufFirst, bufSecond)).writerOffset(size).slice();
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void copyIntoCompositeOffHeapOffHeapBufSlice(Fixture fixture) {
+        try (var a = BufferAllocator.direct();
+             var b = BufferAllocator.direct();
+             var scope = new Scope()) {
+            testCopyIntoBuf(fixture, size -> {
+                int first = size / 2;
+                int second = size - first;
+                try (var bufFirst = a.allocate(first);
+                     var bufSecond = b.allocate(second)) {
+                    return scope.add(Buffer.compose(a, bufFirst, bufSecond)).writerOffset(size).slice();
+                }
+            });
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void byteIterationOfBigEndianBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(0x28)) {
+            buf.order(BIG_ENDIAN); // The byte order should have no impact.
+            checkByteIteration(buf);
+            buf.reset();
+            checkByteIterationOfRegion(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void byteIterationOfLittleEndianBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(0x28)) {
+            buf.order(LITTLE_ENDIAN); // The byte order should have no impact.
+            checkByteIteration(buf);
+            buf.reset();
+            checkByteIterationOfRegion(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void reverseByteIterationOfBigEndianBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(0x28)) {
+            buf.order(BIG_ENDIAN); // The byte order should have no impact.
+            checkReverseByteIteration(buf);
+            buf.reset();
+            checkReverseByteIterationOfRegion(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void reverseByteIterationOfLittleEndianBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(0x28)) {
+            buf.order(LITTLE_ENDIAN); // The byte order should have no impact.
+            checkReverseByteIteration(buf);
+            buf.reset();
+            checkReverseByteIterationOfRegion(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("heapAllocators")
+    public void heapBufferMustHaveZeroAddress(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.nativeAddress()).isZero();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("directAllocators")
+    public void directBufferMustHaveNonZeroAddress(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.nativeAddress()).isNotZero();
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferCleanerTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import io.netty.buffer.api.memseg.NativeMemorySegmentManager;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class BufferCleanerTest extends BufferTestSupport {
+    @Disabled("Too slow, for now")
+    @ParameterizedTest
+    @MethodSource("directAllocators")
+    public void bufferMustBeClosedByCleaner(Fixture fixture) throws InterruptedException {
+        var initial = NativeMemorySegmentManager.MEM_USAGE_NATIVE.sum();
+        var allocator = fixture.createAllocator();
+        allocator.close();
+        int iterations = 15;
+        int allocationSize = 1024;
+        for (int i = 0; i < iterations; i++) {
+            allocateAndForget(allocator, allocationSize);
+            System.gc();
+        }
+        System.runFinalization();
+        var sum = NativeMemorySegmentManager.MEM_USAGE_NATIVE.sum() - initial;
+        var totalAllocated = (long) allocationSize * iterations;
+        assertThat(sum).isLessThan(totalAllocated);
+    }
+
+    private static void allocateAndForget(BufferAllocator allocator, int size) {
+        allocator.allocate(size);
+    }
+
+    @Disabled("Too slow, for now")
+    @ParameterizedTest
+    @MethodSource("pooledDirectAllocators")
+    public void buffersMustBeReusedByPoolingAllocatorEvenWhenDroppedByCleanerInsteadOfExplicitly(Fixture fixture)
+            throws InterruptedException {
+        var initial = NativeMemorySegmentManager.MEM_USAGE_NATIVE.sum();
+        try (var allocator = fixture.createAllocator()) {
+            int iterations = 15;
+            int allocationSize = 1024;
+            for (int i = 0; i < iterations; i++) {
+                allocateAndForget(allocator, allocationSize);
+                System.gc();
+            }
+            System.runFinalization();
+            var sum = NativeMemorySegmentManager.MEM_USAGE_NATIVE.sum() - initial;
+            var totalAllocated = (long) allocationSize * iterations;
+            assertThat(sum).isLessThan(totalAllocated);
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferCompactTest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class BufferCompactTest extends BufferTestSupport {
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void compactMustDiscardReadBytes(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16, BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L).writeInt(0x090A0B0C);
+            assertEquals(0x01020304, buf.readInt());
+            assertEquals(12, buf.writerOffset());
+            assertEquals(4, buf.readerOffset());
+            assertEquals(4, buf.writableBytes());
+            assertEquals(8, buf.readableBytes());
+            assertEquals(16, buf.capacity());
+            buf.compact();
+            assertEquals(8, buf.writerOffset());
+            assertEquals(0, buf.readerOffset());
+            assertEquals(8, buf.writableBytes());
+            assertEquals(8, buf.readableBytes());
+            assertEquals(16, buf.capacity());
+            assertEquals(0x05060708090A0B0CL, buf.readLong());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void compactMustThrowForUnownedBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8, BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            assertEquals((byte) 0x01, buf.readByte());
+            try (Buffer ignore = buf.acquire()) {
+                assertThrows(IllegalStateException.class, () -> buf.compact());
+                assertEquals(1, buf.readerOffset());
+            }
+            assertEquals((byte) 0x02, buf.readByte());
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferComponentIterationTest.java

@@ -0,0 +1,393 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteBuffer;
+import java.nio.ReadOnlyBufferException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class BufferComponentIterationTest extends BufferTestSupport {
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void componentCountOfNonCompositeBufferMustBeOne(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.countComponents()).isOne();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void readableComponentCountMustBeOneIfThereAreReadableBytes(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.countReadableComponents()).isZero();
+            buf.writeByte((byte) 1);
+            assertThat(buf.countReadableComponents()).isOne();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void writableComponentCountMustBeOneIfThereAreWritableBytes(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.countWritableComponents()).isOne();
+            buf.writeLong(1);
+            assertThat(buf.countWritableComponents()).isZero();
+        }
+    }
+
+    @Test
+    public void compositeBufferComponentCountMustBeTransitiveSum() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer buf;
+            try (Buffer a = allocator.allocate(8);
+                 Buffer b = allocator.allocate(8);
+                 Buffer c = allocator.allocate(8);
+                 Buffer x = Buffer.compose(allocator, b, c)) {
+                buf = Buffer.compose(allocator, a, x);
+            }
+            assertThat(buf.countComponents()).isEqualTo(3);
+            assertThat(buf.countReadableComponents()).isZero();
+            assertThat(buf.countWritableComponents()).isEqualTo(3);
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isOne();
+            assertThat(buf.countWritableComponents()).isEqualTo(3);
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isOne();
+            assertThat(buf.countWritableComponents()).isEqualTo(2);
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isEqualTo(2);
+            assertThat(buf.countWritableComponents()).isEqualTo(2);
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isEqualTo(2);
+            assertThat(buf.countWritableComponents()).isOne();
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isEqualTo(3);
+            assertThat(buf.countWritableComponents()).isOne();
+            buf.writeInt(1);
+            assertThat(buf.countReadableComponents()).isEqualTo(3);
+            assertThat(buf.countWritableComponents()).isZero();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void forEachReadableMustVisitBuffer(Fixture fixture) {
+        long value = 0x0102030405060708L;
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer bufBERW = allocator.allocate(8).order(BIG_ENDIAN).writeLong(value);
+             Buffer bufLERW = allocator.allocate(8).order(LITTLE_ENDIAN).writeLong(value);
+             Buffer bufBERO = allocator.allocate(8).order(BIG_ENDIAN).writeLong(value).readOnly(true);
+             Buffer bufLERO = allocator.allocate(8).order(LITTLE_ENDIAN).writeLong(value).readOnly(true)) {
+            verifyForEachReadableSingleComponent(fixture, bufBERW);
+            verifyForEachReadableSingleComponent(fixture, bufLERW);
+            verifyForEachReadableSingleComponent(fixture, bufBERO);
+            verifyForEachReadableSingleComponent(fixture, bufLERO);
+        }
+    }
+
+    @Test
+    public void forEachReadableMustVisitAllReadableConstituentBuffersInOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(4);
+                 Buffer b = allocator.allocate(4);
+                 Buffer c = allocator.allocate(4)) {
+                a.writeInt(1);
+                b.writeInt(2);
+                c.writeInt(3);
+                composite = Buffer.compose(allocator, a, b, c);
+            }
+            var list = new LinkedList<Integer>(List.of(1, 2, 3));
+            int count = composite.forEachReadable(0, (index, component) -> {
+                var buffer = component.readableBuffer();
+                int bufferValue = buffer.getInt();
+                assertEquals(list.pollFirst().intValue(), bufferValue);
+                assertEquals(bufferValue, index + 1);
+                assertThrows(ReadOnlyBufferException.class, () -> buffer.put(0, (byte) 0xFF));
+                return true;
+            });
+            assertEquals(3, count);
+            assertThat(list).isEmpty();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachReadableMustReturnNegativeCountWhenProcessorReturnsFalse(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeLong(0x0102030405060708L);
+            int count = buf.forEachReadable(0, (index, component) -> false);
+            assertEquals(-1, count);
+        }
+    }
+
+    @Test
+    public void forEachReadableMustStopIterationWhenProcessorReturnsFalse() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(4);
+                 Buffer b = allocator.allocate(4);
+                 Buffer c = allocator.allocate(4)) {
+                a.writeInt(1);
+                b.writeInt(2);
+                c.writeInt(3);
+                composite = Buffer.compose(allocator, a, b, c);
+            }
+            int readPos = composite.readerOffset();
+            int writePos = composite.writerOffset();
+            var list = new LinkedList<Integer>(List.of(1, 2, 3));
+            int count = composite.forEachReadable(0, (index, component) -> {
+                var buffer = component.readableBuffer();
+                int bufferValue = buffer.getInt();
+                assertEquals(list.pollFirst().intValue(), bufferValue);
+                assertEquals(bufferValue, index + 1);
+                return false;
+            });
+            assertEquals(-1, count);
+            assertThat(list).containsExactly(2, 3);
+            assertEquals(readPos, composite.readerOffset());
+            assertEquals(writePos, composite.writerOffset());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachReadableOnClosedBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            var buf = allocator.allocate(8);
+            buf.writeLong(0);
+            buf.close();
+            assertThrows(IllegalStateException.class, () -> buf.forEachReadable(0, (component, index) -> true));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachReadableMustAllowCollectingBuffersInArray(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer buf;
+            try (Buffer a = allocator.allocate(4);
+                 Buffer b = allocator.allocate(4);
+                 Buffer c = allocator.allocate(4)) {
+                buf = Buffer.compose(allocator, a, b, c);
+            }
+            int i = 1;
+            while (buf.writableBytes() > 0) {
+                buf.writeByte((byte) i++);
+            }
+            ByteBuffer[] buffers = new ByteBuffer[buf.countReadableComponents()];
+            buf.forEachReadable(0, (index, component) -> {
+                buffers[index] = component.readableBuffer();
+                return true;
+            });
+            i = 1;
+            assertThat(buffers.length).isGreaterThanOrEqualTo(1);
+            for (ByteBuffer buffer : buffers) {
+                while (buffer.hasRemaining()) {
+                    assertEquals((byte) i++, buffer.get());
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachReadableMustExposeByteCursors(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(32).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            buf.writeLong(0x1112131415161718L);
+            assertEquals(0x01020304, buf.readInt());
+            try (Buffer actualData = allocator.allocate(buf.readableBytes()).order(BIG_ENDIAN);
+                 Buffer expectedData = allocator.allocate(12).order(BIG_ENDIAN)) {
+                expectedData.writeInt(0x05060708);
+                expectedData.writeInt(0x11121314);
+                expectedData.writeInt(0x15161718);
+
+                buf.forEachReadable(0, (i, component) -> {
+                    ByteCursor forward = component.openCursor();
+                    while (forward.readLong()) {
+                        actualData.writeLong(forward.getLong());
+                    }
+                    while (forward.readByte()) {
+                        actualData.writeByte(forward.getByte());
+                    }
+                    return true;
+                });
+
+                assertEquals(expectedData.readableBytes(), actualData.readableBytes());
+                while (expectedData.readableBytes() > 0) {
+                    assertEquals(expectedData.readByte(), actualData.readByte());
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void forEachWritableMustVisitBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer bufBERW = allocator.allocate(8).order(BIG_ENDIAN);
+             Buffer bufLERW = allocator.allocate(8).order(LITTLE_ENDIAN)) {
+            verifyForEachWritableSingleComponent(fixture, bufBERW);
+            verifyForEachWritableSingleComponent(fixture, bufLERW);
+        }
+    }
+
+    @Test
+    public void forEachWritableMustVisitAllWritableConstituentBuffersInOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer buf;
+            try (Buffer a = allocator.allocate(8);
+                 Buffer b = allocator.allocate(8);
+                 Buffer c = allocator.allocate(8)) {
+                buf = Buffer.compose(allocator, a, b, c);
+            }
+            buf.order(BIG_ENDIAN);
+            buf.forEachWritable(0, (index, component) -> {
+                component.writableBuffer().putLong(0x0102030405060708L + 0x1010101010101010L * index);
+                return true;
+            });
+            buf.writerOffset(3 * 8);
+            assertEquals(0x0102030405060708L, buf.readLong());
+            assertEquals(0x1112131415161718L, buf.readLong());
+            assertEquals(0x2122232425262728L, buf.readLong());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableMustReturnNegativeCountWhenProcessorReturnsFalse(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int count = buf.forEachWritable(0, (index, component) -> false);
+            assertEquals(-1, count);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableMustStopIterationWhenProcessorRetursFalse(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            AtomicInteger counter = new AtomicInteger();
+            buf.forEachWritable(0, (index, component) -> {
+                counter.incrementAndGet();
+                return false;
+            });
+            assertEquals(1, counter.get());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableChangesMadeToByteBufferComponentMustBeReflectedInBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(9).order(BIG_ENDIAN)) {
+            buf.writeByte((byte) 0xFF);
+            AtomicInteger writtenCounter = new AtomicInteger();
+            buf.forEachWritable(0, (index, component) -> {
+                var buffer = component.writableBuffer();
+                while (buffer.hasRemaining()) {
+                    buffer.put((byte) writtenCounter.incrementAndGet());
+                }
+                return true;
+            });
+            buf.writerOffset(9);
+            assertEquals((byte) 0xFF, buf.readByte());
+            assertEquals(0x0102030405060708L, buf.readLong());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void changesMadeToByteBufferComponentsShouldBeReflectedInBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            AtomicInteger counter = new AtomicInteger();
+            buf.forEachWritable(0, (index, component) -> {
+                var buffer = component.writableBuffer();
+                while (buffer.hasRemaining()) {
+                    buffer.put((byte) counter.incrementAndGet());
+                }
+                return true;
+            });
+            buf.writerOffset(buf.capacity());
+            for (int i = 0; i < 8; i++) {
+                assertEquals((byte) i + 1, buf.getByte(i));
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableOnClosedBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer buf = allocator.allocate(8);
+            buf.close();
+            assertThrows(IllegalStateException.class, () -> buf.forEachWritable(0, (index, component) -> true));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableOnReadOnlyBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8).readOnly(true)) {
+            assertThrows(IllegalStateException.class, () -> buf.forEachWritable(0, (index, component) -> true));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachWritableMustAllowCollectingBuffersInArray(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            ByteBuffer[] buffers = new ByteBuffer[buf.countWritableComponents()];
+            buf.forEachWritable(0, (index, component) -> {
+                buffers[index] = component.writableBuffer();
+                return true;
+            });
+            assertThat(buffers.length).isGreaterThanOrEqualTo(1);
+            int i = 1;
+            for (ByteBuffer buffer : buffers) {
+                while (buffer.hasRemaining()) {
+                    buffer.put((byte) i++);
+                }
+            }
+            buf.writerOffset(buf.capacity());
+            i = 1;
+            while (buf.readableBytes() > 0) {
+                assertEquals((byte) i++, buf.readByte());
+            }
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferCompositionTest.java

@@ -0,0 +1,468 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteOrder;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BufferCompositionTest extends BufferTestSupport {
+
+    @Test
+    public void compositeBufferCanOnlyBeOwnedWhenAllConstituentBuffersAreOwned() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                assertTrue(a.isOwned());
+                Buffer leakB;
+                try (Buffer b = allocator.allocate(8)) {
+                    assertTrue(a.isOwned());
+                    assertTrue(b.isOwned());
+                    composite = Buffer.compose(allocator, a, b);
+                    assertFalse(composite.isOwned());
+                    assertFalse(a.isOwned());
+                    assertFalse(b.isOwned());
+                    leakB = b;
+                }
+                assertFalse(composite.isOwned());
+                assertFalse(a.isOwned());
+                assertTrue(leakB.isOwned());
+            }
+            assertTrue(composite.isOwned());
+        }
+    }
+
+    @Test
+    public void compositeBuffersCannotHaveDuplicateComponents() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(4)) {
+            var e = assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, a, a));
+            assertThat(e).hasMessageContaining("duplicate");
+
+            try (Buffer composite = Buffer.compose(allocator, a)) {
+                a.close();
+                try {
+                    e = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, a));
+                    assertThat(e).hasMessageContaining("duplicate");
+                } finally {
+                    a.acquire();
+                }
+            }
+        }
+    }
+
+    @Test
+    public void compositeBufferFromSends() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator,
+                     allocator.allocate(8).send(),
+                     allocator.allocate(8).send(),
+                     allocator.allocate(8).send())) {
+            assertEquals(24, composite.capacity());
+            assertTrue(composite.isOwned());
+        }
+    }
+
+    @Test
+    public void compositeBufferMustNotBeAllowedToContainThemselves() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer a = allocator.allocate(4);
+            Buffer buf = Buffer.compose(allocator, a);
+            try (buf; a) {
+                a.close();
+                try {
+                    assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(buf, buf));
+                    assertTrue(buf.isOwned());
+                    try (Buffer composite = Buffer.compose(allocator, buf)) {
+                        // the composing increments the reference count of constituent buffers...
+                        // counter-act this so it can be extended:
+                        a.close(); // buf is now owned so it can be extended.
+                        try {
+                            assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(buf, composite));
+                        } finally {
+                            a.acquire(); // restore the reference count to align with our try-with-resources structure.
+                        }
+                    }
+                    assertTrue(buf.isOwned());
+                } finally {
+                    a.acquire();
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableOnCompositeBuffersMustRespectExistingBigEndianByteOrder(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(4, BIG_ENDIAN)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeInt(0x01020304);
+                composite.ensureWritable(4);
+                composite.writeInt(0x05060708);
+                assertEquals(0x0102030405060708L, composite.readLong());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableOnCompositeBuffersMustRespectExistingLittleEndianByteOrder(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(4, LITTLE_ENDIAN)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeInt(0x05060708);
+                composite.ensureWritable(4);
+                composite.writeInt(0x01020304);
+                assertEquals(0x0102030405060708L, composite.readLong());
+            }
+        }
+    }
+
+    @Test
+    public void emptyCompositeBufferMustUseNativeByteOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator)) {
+            assertThat(composite.order()).isEqualTo(ByteOrder.nativeOrder());
+        }
+    }
+
+    @Test
+    public void extendOnNonCompositeBufferMustThrow() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(8);
+             Buffer b = allocator.allocate(8)) {
+            var exc = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(a, b));
+            assertThat(exc).hasMessageContaining("Expected").hasMessageContaining("composite");
+        }
+    }
+
+    @Test
+    public void extendingNonOwnedCompositeBufferMustThrow() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(8);
+             Buffer b = allocator.allocate(8);
+             Buffer composed = Buffer.compose(allocator, a)) {
+            try (Buffer ignore = composed.acquire()) {
+                var exc = assertThrows(IllegalStateException.class, () -> Buffer.extendComposite(composed, b));
+                assertThat(exc).hasMessageContaining("owned");
+            }
+        }
+    }
+
+    @Test
+    public void extendingCompositeBufferWithItselfMustThrow() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                var exc = assertThrows(IllegalArgumentException.class,
+                        () -> Buffer.extendComposite(composite, composite));
+                assertThat(exc).hasMessageContaining("cannot be extended");
+            }
+        }
+    }
+
+    @Test
+    public void extendingWithZeroCapacityBufferHasNoEffect() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator)) {
+            Buffer.extendComposite(composite, composite);
+            assertThat(composite.capacity()).isZero();
+            assertThat(composite.countComponents()).isZero();
+        }
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer a = allocator.allocate(1);
+            Buffer composite = Buffer.compose(allocator, a);
+            a.close();
+            assertTrue(composite.isOwned());
+            assertThat(composite.capacity()).isOne();
+            assertThat(composite.countComponents()).isOne();
+            try (Buffer b = Buffer.compose(allocator)) {
+                Buffer.extendComposite(composite, b);
+            }
+            assertTrue(composite.isOwned());
+            assertThat(composite.capacity()).isOne();
+            assertThat(composite.countComponents()).isOne();
+        }
+    }
+
+    @Test
+    public void extendingCompositeBufferWithNullMustThrow() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator)) {
+            assertThrows(NullPointerException.class, () -> Buffer.extendComposite(composite, null));
+        }
+    }
+
+    @Test
+    public void extendingCompositeBufferMustIncreaseCapacityByGivenBigEndianBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator)) {
+            assertThat(composite.capacity()).isZero();
+            try (Buffer buf = allocator.allocate(8, BIG_ENDIAN)) {
+                Buffer.extendComposite(composite, buf);
+            }
+            assertThat(composite.capacity()).isEqualTo(8);
+            composite.writeLong(0x0102030405060708L);
+            assertThat(composite.readLong()).isEqualTo(0x0102030405060708L);
+        }
+    }
+
+    @Test
+    public void extendingCompositeBufferMustIncreaseCapacityByGivenLittleEndianBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer composite = Buffer.compose(allocator)) {
+            assertThat(composite.capacity()).isZero();
+            try (Buffer buf = allocator.allocate(8, LITTLE_ENDIAN)) {
+                Buffer.extendComposite(composite, buf);
+            }
+            assertThat(composite.capacity()).isEqualTo(8);
+            composite.writeLong(0x0102030405060708L);
+            assertThat(composite.readLong()).isEqualTo(0x0102030405060708L);
+        }
+    }
+
+    @Test
+    public void extendingBigEndianCompositeBufferMustThrowIfExtensionIsLittleEndian() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8, BIG_ENDIAN)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                try (Buffer b = allocator.allocate(8, LITTLE_ENDIAN)) {
+                    var exc = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+                    assertThat(exc).hasMessageContaining("byte order");
+                }
+            }
+        }
+    }
+
+    @Test
+    public void extendingLittleEndianCompositeBufferMustThrowIfExtensionIsBigEndian() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8, LITTLE_ENDIAN)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                try (Buffer b = allocator.allocate(8, BIG_ENDIAN)) {
+                    var exc = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+                    assertThat(exc).hasMessageContaining("byte order");
+                }
+            }
+        }
+    }
+
+    @Test
+    public void emptyCompositeBufferMustAllowExtendingWithBufferWithBigEndianByteOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            try (Buffer composite = Buffer.compose(allocator)) {
+                try (Buffer b = allocator.allocate(8, BIG_ENDIAN)) {
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.order()).isEqualTo(BIG_ENDIAN);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void emptyCompositeBufferMustAllowExtendingWithBufferWithLittleEndianByteOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            try (Buffer composite = Buffer.compose(allocator)) {
+                try (Buffer b = allocator.allocate(8, LITTLE_ENDIAN)) {
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.order()).isEqualTo(LITTLE_ENDIAN);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void emptyCompositeBufferMustAllowExtendingWithReadOnlyBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            try (Buffer composite = Buffer.compose(allocator)) {
+                try (Buffer b = allocator.allocate(8).readOnly(true)) {
+                    Buffer.extendComposite(composite, b);
+                    assertTrue(composite.readOnly());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void whenExtendingCompositeBufferWithWriteOffsetAtCapacityExtensionWriteOffsetCanBeNonZero() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeLong(0);
+                try (Buffer b = allocator.allocate(8)) {
+                    b.writeInt(1);
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.capacity()).isEqualTo(16);
+                    assertThat(composite.writerOffset()).isEqualTo(12);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void whenExtendingCompositeBufferWithWriteOffsetLessThanCapacityExtensionWriteOffsetMustZero() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeInt(0);
+                try (Buffer b = allocator.allocate(8)) {
+                    b.writeInt(1);
+                    var exc = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+                    assertThat(exc).hasMessageContaining("unwritten gap");
+                    b.writerOffset(0);
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.capacity()).isEqualTo(16);
+                    assertThat(composite.writerOffset()).isEqualTo(4);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void whenExtendingCompositeBufferWithReadOffsetAtCapacityExtensionReadOffsetCanBeNonZero() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeLong(0);
+                composite.readLong();
+                try (Buffer b = allocator.allocate(8)) {
+                    b.writeInt(1);
+                    b.readInt();
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.capacity()).isEqualTo(16);
+                    assertThat(composite.writerOffset()).isEqualTo(12);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void whenExtendingCompositeBufferWithReadOffsetLessThanCapacityExtensionReadOffsetMustZero() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite) {
+                composite.writeLong(0);
+                composite.readInt();
+                try (Buffer b = allocator.allocate(8)) {
+                    b.writeInt(1);
+                    b.readInt();
+                    var exc = assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+                    assertThat(exc).hasMessageContaining("unread gap");
+                    b.readerOffset(0);
+                    Buffer.extendComposite(composite, b);
+                    assertThat(composite.capacity()).isEqualTo(16);
+                    assertThat(composite.writerOffset()).isEqualTo(12);
+                    assertThat(composite.readerOffset()).isEqualTo(4);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void composeMustThrowWhenBuffersHaveMismatchedByteOrder() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(4, BIG_ENDIAN);
+             Buffer b = allocator.allocate(4, LITTLE_ENDIAN)) {
+            assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, a, b));
+        }
+    }
+
+    @Test
+    public void composingReadOnlyBuffersMustCreateReadOnlyCompositeBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(4).readOnly(true);
+             Buffer b = allocator.allocate(4).readOnly(true);
+             Buffer composite = Buffer.compose(allocator, a, b)) {
+            assertTrue(composite.readOnly());
+            verifyWriteInaccessible(composite);
+        }
+    }
+
+    @Test
+    public void composingReadOnlyAndWritableBuffersMustThrow() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer a = allocator.allocate(8).readOnly(true);
+             Buffer b = allocator.allocate(8)) {
+            assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, a, b));
+            assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, b, a));
+            assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, a, b, a));
+            assertThrows(IllegalArgumentException.class, () -> Buffer.compose(allocator, b, a, b));
+        }
+    }
+
+    @Test
+    public void compositeWritableBufferCannotBeExtendedWithReadOnlyBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite; Buffer b = allocator.allocate(8).readOnly(true)) {
+                assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+            }
+        }
+    }
+
+    @Test
+    public void compositeReadOnlyBufferCannotBeExtendedWithWritableBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap()) {
+            Buffer composite;
+            try (Buffer a = allocator.allocate(8).readOnly(true)) {
+                composite = Buffer.compose(allocator, a);
+            }
+            try (composite; Buffer b = allocator.allocate(8)) {
+                assertThrows(IllegalArgumentException.class, () -> Buffer.extendComposite(composite, b));
+            }
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferEnsureWritableTest.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class BufferEnsureWritableTest extends BufferTestSupport {
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableMustThrowForBorrowedBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            try (Buffer slice = buf.slice()) {
+                assertThrows(IllegalStateException.class, () -> slice.ensureWritable(1));
+                assertThrows(IllegalStateException.class, () -> buf.ensureWritable(1));
+            }
+            try (Buffer compose = Buffer.compose(allocator, buf)) {
+                assertThrows(IllegalStateException.class, () -> compose.ensureWritable(1));
+                assertThrows(IllegalStateException.class, () -> buf.ensureWritable(1));
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableMustThrowForNegativeSize(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThrows(IllegalArgumentException.class, () -> buf.ensureWritable(-1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableMustThrowIfRequestedSizeWouldGrowBeyondMaxAllowed(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(512)) {
+            assertThrows(IllegalArgumentException.class, () -> buf.ensureWritable(Integer.MAX_VALUE - 8));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableMustNotThrowWhenSpaceIsAlreadyAvailable(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.ensureWritable(8);
+            buf.writeLong(1);
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.writeByte((byte) 1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableMustExpandBufferCapacity(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.writableBytes()).isEqualTo(8);
+            buf.writeLong(0x0102030405060708L);
+            assertThat(buf.writableBytes()).isEqualTo(0);
+            buf.ensureWritable(8);
+            assertThat(buf.writableBytes()).isGreaterThanOrEqualTo(8);
+            assertThat(buf.capacity()).isGreaterThanOrEqualTo(16);
+            buf.writeLong(0xA1A2A3A4A5A6A7A8L);
+            assertThat(buf.readableBytes()).isEqualTo(16);
+            assertThat(buf.readLong()).isEqualTo(0x0102030405060708L);
+            assertThat(buf.readLong()).isEqualTo(0xA1A2A3A4A5A6A7A8L);
+            assertThrows(IndexOutOfBoundsException.class, buf::readByte);
+            // Is it implementation dependent if the capacity increased by *exactly* the requested size, or more.
+        }
+    }
+
+    @Test
+    public void ensureWritableMustExpandCapacityOfEmptyCompositeBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer buf = Buffer.compose(allocator)) {
+            assertThat(buf.writableBytes()).isEqualTo(0);
+            buf.ensureWritable(8);
+            assertThat(buf.writableBytes()).isGreaterThanOrEqualTo(8);
+            buf.writeLong(0xA1A2A3A4A5A6A7A8L);
+            assertThat(buf.readableBytes()).isEqualTo(8);
+            assertThat(buf.readLong()).isEqualTo(0xA1A2A3A4A5A6A7A8L);
+            assertThrows(IndexOutOfBoundsException.class, buf::readByte);
+            // Is it implementation dependent if the capacity increased by *exactly* the requested size, or more.
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void mustBeAbleToSliceAfterEnsureWritable(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(4)) {
+            buf.ensureWritable(8);
+            assertThat(buf.writableBytes()).isGreaterThanOrEqualTo(8);
+            assertThat(buf.capacity()).isGreaterThanOrEqualTo(8);
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer slice = buf.slice()) {
+                assertEquals(0x0102030405060708L, slice.readLong());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableWithCompactionMustNotAllocateIfCompactionIsEnough(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(64)) {
+            while (buf.writableBytes() > 0) {
+                buf.writeByte((byte) 42);
+            }
+            while (buf.readableBytes() > 0) {
+                buf.readByte();
+            }
+            buf.ensureWritable(4, true);
+            buf.writeInt(42);
+            assertThat(buf.capacity()).isEqualTo(64);
+
+            buf.writerOffset(60).readerOffset(60);
+            buf.ensureWritable(8, true);
+            buf.writeLong(42);
+            // Don't assert the capacity on this one, because single-component
+            // composite buffers may choose to allocate rather than compact.
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferOffsetsTest.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class BufferOffsetsTest extends BufferTestSupport {
+
+    @ParameterizedTest
+    @MethodSource("initialAllocators")
+    void mustThrowWhenAllocatingZeroSizedBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            assertThrows(IllegalArgumentException.class, () -> allocator.allocate(0));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void mustThrowWhenAllocatingNegativeSizedBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            assertThrows(IllegalArgumentException.class, () -> allocator.allocate(-1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void setReaderOffsetMustThrowOnNegativeIndex(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.readerOffset(-1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void setReaderOffsetMustThrowOnOversizedIndex(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.readerOffset(1));
+            buf.writeLong(0);
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.readerOffset(9));
+
+            buf.readerOffset(8);
+            assertThrows(IndexOutOfBoundsException.class, buf::readByte);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void setWriterOffsetMustThrowOutsideOfWritableRegion(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            // Writer offset cannot be negative.
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.writerOffset(-1));
+
+            buf.writerOffset(4);
+            buf.readerOffset(4);
+
+            // Cannot set writer offset before reader offset.
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.writerOffset(3));
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.writerOffset(0));
+
+            buf.writerOffset(buf.capacity());
+
+            // Cannot set writer offset beyond capacity.
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.writerOffset(buf.capacity() + 1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void setReaderOffsetMustNotThrowWithinBounds(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.readerOffset(0)).isSameAs(buf);
+            buf.writeLong(0);
+            assertThat(buf.readerOffset(7)).isSameAs(buf);
+            assertThat(buf.readerOffset(8)).isSameAs(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void capacityMustBeAllocatedSize(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertEquals(8, buf.capacity());
+            try (Buffer b = allocator.allocate(13)) {
+                assertEquals(13, b.capacity());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void readerWriterOffsetUpdates(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(22)) {
+            assertEquals(0, buf.writerOffset());
+            assertThat(buf.writerOffset(1)).isSameAs(buf);
+            assertEquals(1, buf.writerOffset());
+            assertThat(buf.writeByte((byte) 7)).isSameAs(buf);
+            assertEquals(2, buf.writerOffset());
+            assertThat(buf.writeShort((short) 3003)).isSameAs(buf);
+            assertEquals(4, buf.writerOffset());
+            assertThat(buf.writeInt(0x5A55_BA55)).isSameAs(buf);
+            assertEquals(8, buf.writerOffset());
+            assertThat(buf.writeLong(0x123456789ABCDEF0L)).isSameAs(buf);
+            assertEquals(16, buf.writerOffset());
+            assertEquals(6, buf.writableBytes());
+            assertEquals(16, buf.readableBytes());
+
+            assertEquals(0, buf.readerOffset());
+            assertThat(buf.readerOffset(1)).isSameAs(buf);
+            assertEquals(1, buf.readerOffset());
+            assertEquals((byte) 7, buf.readByte());
+            assertEquals(2, buf.readerOffset());
+            assertEquals((short) 3003, buf.readShort());
+            assertEquals(4, buf.readerOffset());
+            assertEquals(0x5A55_BA55, buf.readInt());
+            assertEquals(8, buf.readerOffset());
+            assertEquals(0x123456789ABCDEF0L, buf.readLong());
+            assertEquals(16, buf.readerOffset());
+            assertEquals(0, buf.readableBytes());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void readAndWriteBoundsChecksWithIndexUpdates(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeLong(0);
+
+            buf.readLong(); // Fine.
+            buf.readerOffset(1);
+            assertThrows(IndexOutOfBoundsException.class, buf::readLong);
+
+            buf.readerOffset(4);
+            buf.readInt(); // Fine.
+            buf.readerOffset(5);
+
+            assertThrows(IndexOutOfBoundsException.class, buf::readInt);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void resetMustSetReaderAndWriterOffsetsToTheirInitialPositions(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeInt(0).readShort();
+            buf.reset();
+            assertEquals(0, buf.readerOffset());
+            assertEquals(0, buf.writerOffset());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void readableBytesMustMatchWhatWasWritten(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16)) {
+            buf.writeLong(0);
+            assertEquals(Long.BYTES, buf.readableBytes());
+            buf.readShort();
+            assertEquals(Long.BYTES - Short.BYTES, buf.readableBytes());
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferReadOnlyTest.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BufferReadOnlyTest extends BufferTestSupport {
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void readOnlyBufferMustPreventWriteAccess(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            var b = buf.readOnly(true);
+            assertThat(b).isSameAs(buf);
+            verifyWriteInaccessible(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void closedBuffersAreNotReadOnly(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer buf = allocator.allocate(8);
+            buf.readOnly(true);
+            buf.close();
+            assertFalse(buf.readOnly());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void readOnlyBufferMustBecomeWritableAgainAfterTogglingReadOnlyOff(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertFalse(buf.readOnly());
+            buf.readOnly(true);
+            assertTrue(buf.readOnly());
+            verifyWriteInaccessible(buf);
+
+            buf.readOnly(false);
+            assertFalse(buf.readOnly());
+
+            verifyWriteAccessible(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void readOnlyBufferMustRemainReadOnlyAfterSend(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.readOnly(true);
+            var send = buf.send();
+            try (Buffer receive = send.receive()) {
+                assertTrue(receive.readOnly());
+                verifyWriteInaccessible(receive);
+            }
+        }
+    }
+
+    @Test
+    public void readOnlyBufferMustRemainReadOnlyAfterSendForEmptyCompositeBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer buf = Buffer.compose(allocator)) {
+            buf.readOnly(true);
+            var send = buf.send();
+            try (Buffer receive = send.receive()) {
+                assertTrue(receive.readOnly());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("pooledAllocators")
+    public void readOnlyBufferMustNotBeReadOnlyAfterBeingReusedFromPool(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            for (int i = 0; i < 1000; i++) {
+                try (Buffer buf = allocator.allocate(8)) {
+                    assertFalse(buf.readOnly());
+                    buf.readOnly(true);
+                    assertTrue(buf.readOnly());
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void compactOnReadOnlyBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.readOnly(true);
+            assertThrows(IllegalStateException.class, () -> buf.compact());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableOnReadOnlyBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.readOnly(true);
+            assertThrows(IllegalStateException.class, () -> buf.ensureWritable(1));
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void copyIntoOnReadOnlyBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer dest = allocator.allocate(8)) {
+            dest.readOnly(true);
+            try (Buffer src = allocator.allocate(8)) {
+                assertThrows(IllegalStateException.class, () -> src.copyInto(0, dest, 0, 1));
+            }
+        }
+    }
+    // todo read only buffer must have zero writable bytes
+}

src/test/java/io/netty/buffer/api/BufferReferenceCountingTest.java

@@ -0,0 +1,642 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BufferReferenceCountingTest extends BufferTestSupport {
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void allocateAndAccessingBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeByte((byte) 1);
+            buf.writeByte((byte) 2);
+            try (Buffer inner = buf.acquire()) {
+                inner.writeByte((byte) 3);
+                inner.writeByte((byte) 4);
+                inner.writeByte((byte) 5);
+                inner.writeByte((byte) 6);
+                inner.writeByte((byte) 7);
+                inner.writeByte((byte) 8);
+                var re = assertThrows(RuntimeException.class, () -> inner.writeByte((byte) 9));
+                assertThat(re).hasMessageContaining("bound");
+                re = assertThrows(RuntimeException.class, () -> inner.writeByte((byte) 9));
+                assertThat(re).hasMessageContaining("bound");
+                re = assertThrows(RuntimeException.class, () -> buf.writeByte((byte) 9));
+                assertThat(re).hasMessageContaining("bound");
+            }
+            assertEquals((byte) 1, buf.readByte());
+            assertEquals((byte) 2, buf.readByte());
+            assertEquals((byte) 3, buf.readByte());
+            assertEquals((byte) 4, buf.readByte());
+            assertEquals((byte) 5, buf.readByte());
+            assertEquals((byte) 6, buf.readByte());
+            assertEquals((byte) 7, buf.readByte());
+            assertEquals((byte) 8, buf.readByte());
+            var re = assertThrows(RuntimeException.class, buf::readByte);
+            assertThat(re).hasMessageContaining("bound");
+            assertThat(toByteArray(buf)).containsExactly(1, 2, 3, 4, 5, 6, 7, 8);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void acquireOnClosedBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            var buf = allocator.allocate(8);
+            buf.close();
+            assertThrows(IllegalStateException.class, buf::acquire);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bufferShouldNotBeAccessibleAfterClose(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer buf = allocator.allocate(24);
+            buf.writeLong(42);
+            buf.close();
+            verifyInaccessible(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bufferMustNotBeThreadConfined(Fixture fixture) throws Exception {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeInt(42);
+            Future<Integer> fut = executor.submit(() -> buf.readInt());
+            assertEquals(42, fut.get());
+            fut = executor.submit(() -> {
+                buf.writeInt(32);
+                return buf.readInt();
+            });
+            assertEquals(32, fut.get());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithoutOffsetAndSizeMustReturnReadableRegion(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            for (byte b : new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08 }) {
+                buf.writeByte(b);
+            }
+            assertEquals(0x01, buf.readByte());
+            buf.writerOffset(buf.writerOffset() - 1);
+            try (Buffer slice = buf.slice()) {
+                assertThat(toByteArray(slice)).containsExactly(0x02, 0x03, 0x04, 0x05, 0x06, 0x07);
+                assertEquals(0, slice.readerOffset());
+                assertEquals(6, slice.readableBytes());
+                assertEquals(6, slice.writerOffset());
+                assertEquals(6, slice.capacity());
+                assertEquals(0x02, slice.readByte());
+                assertEquals(0x03, slice.readByte());
+                assertEquals(0x04, slice.readByte());
+                assertEquals(0x05, slice.readByte());
+                assertEquals(0x06, slice.readByte());
+                assertEquals(0x07, slice.readByte());
+                assertThrows(IndexOutOfBoundsException.class, slice::readByte);
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithOffsetAndSizeMustReturnGivenRegion(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            for (byte b : new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08 }) {
+                buf.writeByte(b);
+            }
+            buf.readerOffset(3); // Reader and writer offsets must be ignored.
+            buf.writerOffset(6);
+            try (Buffer slice = buf.slice(1, 6)) {
+                assertThat(toByteArray(slice)).containsExactly(0x02, 0x03, 0x04, 0x05, 0x06, 0x07);
+                assertEquals(0, slice.readerOffset());
+                assertEquals(6, slice.readableBytes());
+                assertEquals(6, slice.writerOffset());
+                assertEquals(6, slice.capacity());
+                assertEquals(0x02, slice.readByte());
+                assertEquals(0x03, slice.readByte());
+                assertEquals(0x04, slice.readByte());
+                assertEquals(0x05, slice.readByte());
+                assertEquals(0x06, slice.readByte());
+                assertEquals(0x07, slice.readByte());
+                assertThrows(IndexOutOfBoundsException.class, slice::readByte);
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithoutOffsetAndSizeWillIncreaseReferenceCount(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            try (Buffer ignored = buf.slice()) {
+                assertFalse(buf.isOwned());
+                assertThrows(IllegalStateException.class, buf::send);
+            }
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithOffsetAndSizeWillIncreaseReferenceCount(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            try (Buffer ignored = buf.slice(0, 8)) {
+                assertFalse(buf.isOwned());
+                assertThrows(IllegalStateException.class, buf::send);
+            }
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithoutOffsetAndSizeHasSameEndianAsParent(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.order(BIG_ENDIAN);
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer slice = buf.slice()) {
+                assertEquals(0x0102030405060708L, slice.readLong());
+            }
+            buf.order(LITTLE_ENDIAN);
+            try (Buffer slice = buf.slice()) {
+                assertEquals(0x0807060504030201L, slice.readLong());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithOffsetAndSizeHasSameEndianAsParent(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.order(BIG_ENDIAN);
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer slice = buf.slice(0, 8)) {
+                assertEquals(0x0102030405060708L, slice.readLong());
+            }
+            buf.order(LITTLE_ENDIAN);
+            try (Buffer slice = buf.slice(0, 8)) {
+                assertEquals(0x0807060504030201L, slice.readLong());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sendOnSliceWithoutOffsetAndSizeMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            try (Buffer slice = buf.slice()) {
+                assertFalse(buf.isOwned());
+                assertThrows(IllegalStateException.class, slice::send);
+            }
+            // Verify that the slice is closed properly afterwards.
+            assertTrue(buf.isOwned());
+            buf.send().receive().close();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sendOnSliceWithOffsetAndSizeMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            try (Buffer slice = buf.slice(0, 8)) {
+                assertFalse(buf.isOwned());
+                assertThrows(IllegalStateException.class, slice::send);
+            }
+            // Verify that the slice is closed properly afterwards.
+            assertTrue(buf.isOwned());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithNegativeOffsetMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.slice(-1, 1));
+            // Verify that the slice is closed properly afterwards.
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithNegativeSizeMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            assertThrows(IllegalArgumentException.class, () -> buf.slice(0, -1));
+            // Verify that the slice is closed properly afterwards.
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithSizeGreaterThanCapacityMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.slice(0, 9));
+            buf.slice(0, 8).close(); // This is still fine.
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.slice(1, 8));
+            // Verify that the slice is closed properly afterwards.
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sliceWithZeroSizeMustBeAllowed(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            buf.slice(0, 0).close(); // This is fine.
+            // Verify that the slice is closed properly afterwards.
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void acquireComposingAndSlicingMustIncrementBorrows(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            int borrows = buf.countBorrows();
+            try (Buffer ignored = buf.acquire()) {
+                assertEquals(borrows + 1, buf.countBorrows());
+                try (Buffer slice = buf.slice()) {
+                    assertEquals(0, slice.capacity()); // We haven't written anything, so the slice is empty.
+                    int sliceBorrows = slice.countBorrows();
+                    assertEquals(borrows + 2, buf.countBorrows());
+                    try (Buffer ignored1 = Buffer.compose(allocator, buf, slice)) {
+                        assertEquals(borrows + 3, buf.countBorrows());
+                        // Note: Slice is empty; not acquired by the composite buffer.
+                        assertEquals(sliceBorrows, slice.countBorrows());
+                    }
+                    assertEquals(sliceBorrows, slice.countBorrows());
+                    assertEquals(borrows + 2, buf.countBorrows());
+                }
+                assertEquals(borrows + 1, buf.countBorrows());
+            }
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("nonCompositeAllocators")
+    public void acquireComposingAndSlicingMustIncrementBorrowsWithData(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeByte((byte) 1);
+            int borrows = buf.countBorrows();
+            try (Buffer ignored = buf.acquire()) {
+                assertEquals(borrows + 1, buf.countBorrows());
+                try (Buffer slice = buf.slice()) {
+                    assertEquals(1, slice.capacity());
+                    int sliceBorrows = slice.countBorrows();
+                    assertEquals(borrows + 2, buf.countBorrows());
+                    try (Buffer ignored1 = Buffer.compose(allocator, buf, slice)) {
+                        assertEquals(borrows + 3, buf.countBorrows());
+                        assertEquals(sliceBorrows + 1, slice.countBorrows());
+                    }
+                    assertEquals(sliceBorrows, slice.countBorrows());
+                    assertEquals(borrows + 2, buf.countBorrows());
+                }
+                assertEquals(borrows + 1, buf.countBorrows());
+            }
+            assertEquals(borrows, buf.countBorrows());
+        }
+    }
+
+    @Disabled // TODO
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void sliceMustBecomeOwnedOnSourceBufferClose(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            Buffer buf = allocator.allocate(8);
+            buf.writeInt(42);
+            try (Buffer slice = buf.slice()) {
+                buf.close();
+                assertFalse(buf.isAccessible());
+                assertTrue(slice.isOwned());
+                try (Buffer receive = slice.send().receive()) {
+                    assertTrue(receive.isOwned());
+                    assertFalse(slice.isAccessible());
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void pooledBuffersMustResetStateBeforeReuse(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer expected = allocator.allocate(8)) {
+            for (int i = 0; i < 10; i++) {
+                try (Buffer buf = allocator.allocate(8)) {
+                    assertEquals(expected.capacity(), buf.capacity());
+                    assertEquals(expected.readableBytes(), buf.readableBytes());
+                    assertEquals(expected.readerOffset(), buf.readerOffset());
+                    assertEquals(expected.writableBytes(), buf.writableBytes());
+                    assertEquals(expected.writerOffset(), buf.writerOffset());
+                    assertThat(buf.order()).isEqualTo(expected.order());
+                    byte[] bytes = new byte[8];
+                    buf.copyInto(0, bytes, 0, 8);
+                    assertThat(bytes).containsExactly(0, 0, 0, 0, 0, 0, 0, 0);
+
+                    var tlr = ThreadLocalRandom.current();
+                    buf.order(tlr.nextBoolean()? LITTLE_ENDIAN : BIG_ENDIAN);
+                    for (int j = 0; j < tlr.nextInt(0, 8); j++) {
+                        buf.writeByte((byte) 1);
+                    }
+                    if (buf.readableBytes() > 0) {
+                        for (int j = 0; j < tlr.nextInt(0, buf.readableBytes()); j++) {
+                            buf.readByte();
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcateOfNonOwnedBufferMustThrow(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeInt(1);
+            try (Buffer acquired = buf.acquire()) {
+                var exc = assertThrows(IllegalStateException.class, () -> acquired.bifurcate());
+                assertThat(exc).hasMessageContaining("owned");
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcatedPartMustContainFirstHalfOfBuffer(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            assertThat(buf.readByte()).isEqualTo((byte) 0x01);
+            try (Buffer bif = buf.bifurcate()) {
+                // Original buffer:
+                assertThat(buf.capacity()).isEqualTo(8);
+                assertThat(buf.readerOffset()).isZero();
+                assertThat(buf.writerOffset()).isZero();
+                assertThat(buf.readableBytes()).isZero();
+                assertThrows(IndexOutOfBoundsException.class, () -> buf.readByte());
+
+                // Bifurcated part:
+                assertThat(bif.capacity()).isEqualTo(8);
+                assertThat(bif.readerOffset()).isOne();
+                assertThat(bif.writerOffset()).isEqualTo(8);
+                assertThat(bif.readableBytes()).isEqualTo(7);
+                assertThat(bif.readByte()).isEqualTo((byte) 0x02);
+                assertThat(bif.readInt()).isEqualTo(0x03040506);
+                assertThat(bif.readByte()).isEqualTo((byte) 0x07);
+                assertThat(bif.readByte()).isEqualTo((byte) 0x08);
+                assertThrows(IndexOutOfBoundsException.class, () -> bif.readByte());
+            }
+
+            // Bifurcated part does NOT return when closed:
+            assertThat(buf.capacity()).isEqualTo(8);
+            assertThat(buf.readerOffset()).isZero();
+            assertThat(buf.writerOffset()).isZero();
+            assertThat(buf.readableBytes()).isZero();
+            assertThrows(IndexOutOfBoundsException.class, () -> buf.readByte());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcatedPartsMustBeIndividuallySendable(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            assertThat(buf.readByte()).isEqualTo((byte) 0x01);
+            try (Buffer sentBif = buf.bifurcate().send().receive()) {
+                try (Buffer sentBuf = buf.send().receive()) {
+                    assertThat(sentBuf.capacity()).isEqualTo(8);
+                    assertThat(sentBuf.readerOffset()).isZero();
+                    assertThat(sentBuf.writerOffset()).isZero();
+                    assertThat(sentBuf.readableBytes()).isZero();
+                    assertThrows(IndexOutOfBoundsException.class, () -> sentBuf.readByte());
+                }
+
+                assertThat(sentBif.capacity()).isEqualTo(8);
+                assertThat(sentBif.readerOffset()).isOne();
+                assertThat(sentBif.writerOffset()).isEqualTo(8);
+                assertThat(sentBif.readableBytes()).isEqualTo(7);
+                assertThat(sentBif.readByte()).isEqualTo((byte) 0x02);
+                assertThat(sentBif.readInt()).isEqualTo(0x03040506);
+                assertThat(sentBif.readByte()).isEqualTo((byte) 0x07);
+                assertThat(sentBif.readByte()).isEqualTo((byte) 0x08);
+                assertThrows(IndexOutOfBoundsException.class, () -> sentBif.readByte());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void mustBePossibleToBifurcateMoreThanOnce(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer a = buf.bifurcate()) {
+                a.writerOffset(4);
+                try (Buffer b = a.bifurcate()) {
+                    assertEquals(0x01020304, b.readInt());
+                    a.writerOffset(4);
+                    assertEquals(0x05060708, a.readInt());
+                    assertThrows(IndexOutOfBoundsException.class, () -> b.readByte());
+                    assertThrows(IndexOutOfBoundsException.class, () -> a.readByte());
+                    buf.writeLong(0xA1A2A3A4A5A6A7A8L);
+                    buf.writerOffset(4);
+                    try (Buffer c = buf.bifurcate()) {
+                        assertEquals(0xA1A2A3A4, c.readInt());
+                        buf.writerOffset(4);
+                        assertEquals(0xA5A6A7A8, buf.readInt());
+                        assertThrows(IndexOutOfBoundsException.class, () -> c.readByte());
+                        assertThrows(IndexOutOfBoundsException.class, () -> buf.readByte());
+                    }
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcatedBufferMustHaveSameByteOrderAsParent(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer a = buf.bifurcate()) {
+                assertThat(a.order()).isEqualTo(BIG_ENDIAN);
+                a.order(LITTLE_ENDIAN);
+                a.writerOffset(4);
+                try (Buffer b = a.bifurcate()) {
+                    assertThat(b.order()).isEqualTo(LITTLE_ENDIAN);
+                    assertThat(buf.order()).isEqualTo(BIG_ENDIAN);
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableOnBifurcatedBuffers(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeLong(0x0102030405060708L);
+            try (Buffer a = buf.bifurcate()) {
+                assertEquals(0x0102030405060708L, a.readLong());
+                a.ensureWritable(8);
+                a.writeLong(0xA1A2A3A4A5A6A7A8L);
+                assertEquals(0xA1A2A3A4A5A6A7A8L, a.readLong());
+
+                buf.ensureWritable(8);
+                buf.writeLong(0xA1A2A3A4A5A6A7A8L);
+                assertEquals(0xA1A2A3A4A5A6A7A8L, buf.readLong());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void ensureWritableOnBifurcatedBuffersWithOddOffsets(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(10).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            buf.writeByte((byte) 0x09);
+            buf.readByte();
+            try (Buffer a = buf.bifurcate()) {
+                assertEquals(0x0203040506070809L, a.readLong());
+                a.ensureWritable(8);
+                a.writeLong(0xA1A2A3A4A5A6A7A8L);
+                assertEquals(0xA1A2A3A4A5A6A7A8L, a.readLong());
+
+                buf.ensureWritable(8);
+                buf.writeLong(0xA1A2A3A4A5A6A7A8L);
+                assertEquals(0xA1A2A3A4A5A6A7A8L, buf.readLong());
+            }
+        }
+    }
+
+    @Test
+    public void bifurcateOnEmptyBigEndianCompositeBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer buf = Buffer.compose(allocator).order(BIG_ENDIAN)) {
+            verifyBifurcateEmptyCompositeBuffer(buf);
+        }
+    }
+
+    @Test
+    public void bifurcateOnEmptyLittleEndianCompositeBuffer() {
+        try (BufferAllocator allocator = BufferAllocator.heap();
+             Buffer buf = Buffer.compose(allocator).order(LITTLE_ENDIAN)) {
+            verifyBifurcateEmptyCompositeBuffer(buf);
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcatedBuffersMustBeAccessibleInOtherThreads(Fixture fixture) throws Exception {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeInt(42);
+            var send = buf.bifurcate().send();
+            var fut = executor.submit(() -> {
+                try (Buffer receive = send.receive()) {
+                    assertEquals(42, receive.readInt());
+                    receive.readerOffset(0).writerOffset(0).writeInt(24);
+                    assertEquals(24, receive.readInt());
+                }
+            });
+            fut.get();
+            buf.writeInt(32);
+            assertEquals(32, buf.readInt());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void acquireOfReadOnlyBufferMustBeReadOnly(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.readOnly(true);
+            try (Buffer acquire = buf.acquire()) {
+                assertTrue(acquire.readOnly());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void sliceOfReadOnlyBufferMustBeReadOnly(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            buf.writeLong(0x0102030405060708L);
+            buf.readOnly(true);
+            try (Buffer slice = buf.slice()) {
+                assertTrue(slice.readOnly());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void bifurcateOfReadOnlyBufferMustBeReadOnly(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16)) {
+            buf.writeLong(0x0102030405060708L);
+            buf.readOnly(true);
+            try (Buffer bifurcate = buf.bifurcate()) {
+                assertTrue(bifurcate.readOnly());
+                assertTrue(buf.readOnly());
+            }
+        }
+    }
+}

src/test/java/io/netty/buffer/api/BufferSendTest.java

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Future;
+import java.util.concurrent.SynchronousQueue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BufferSendTest extends BufferTestSupport {
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void allocateAndSendToThread(Fixture fixture) throws Exception {
+        try (BufferAllocator allocator = fixture.createAllocator()) {
+            ArrayBlockingQueue<Send<Buffer>> queue = new ArrayBlockingQueue<>(10);
+            Future<Byte> future = executor.submit(() -> {
+                try (Buffer byteBuf = queue.take().receive()) {
+                    return byteBuf.readByte();
+                }
+            });
+
+            try (Buffer buf = allocator.allocate(8)) {
+                buf.writeByte((byte) 42);
+                assertTrue(queue.offer(buf.send()));
+            }
+
+            assertEquals((byte) 42, future.get().byteValue());
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void allocateAndSendToThreadViaSyncQueue(Fixture fixture) throws Exception {
+        SynchronousQueue<Send<Buffer>> queue = new SynchronousQueue<>();
+        Future<Byte> future = executor.submit(() -> {
+            try (Buffer byteBuf = queue.take().receive()) {
+                return byteBuf.readByte();
+            }
+        });
+
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            assertThat(buf.writeByte((byte) 42)).isSameAs(buf);
+            queue.put(buf.send());
+        }
+
+        assertEquals((byte) 42, future.get().byteValue());
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    void sendMustThrowWhenBufIsAcquired(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            try (Buffer ignored = buf.acquire()) {
+                assertFalse(buf.isOwned());
+                assertThrows(IllegalStateException.class, buf::send);
+            }
+            // Now send() should work again.
+            assertTrue(buf.isOwned());
+            buf.send().receive().close();
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void originalBufferMustNotBeAccessibleAfterSend(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer orig = allocator.allocate(24)) {
+            orig.writeLong(42);
+            var send = orig.send();
+            verifyInaccessible(orig);
+            try (Buffer receive = send.receive()) {
+                assertEquals(42, receive.readInt());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void cannotSendMoreThanOnce(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8)) {
+            var send = buf.send();
+            var exc = assertThrows(IllegalStateException.class, () -> buf.send());
+            send.receive().close();
+            assertThat(exc).hasMessageContaining("Cannot send()");
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void sendMustNotMakeBifurcatedBuffersInaccessible(Fixture fixture) throws Exception {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(16)) {
+            buf.writeInt(64);
+            var bifA = buf.bifurcate();
+            buf.writeInt(42);
+            var send = buf.bifurcate().send();
+            buf.writeInt(72);
+            var bifB = buf.bifurcate();
+            var fut = executor.submit(() -> {
+                try (Buffer receive = send.receive()) {
+                    assertEquals(42, receive.readInt());
+                }
+            });
+            fut.get();
+            buf.writeInt(32);
+            assertEquals(32, buf.readInt());
+            assertEquals(64, bifA.readInt());
+            assertEquals(72, bifB.readInt());
+        }
+    }
+}