diff --git a/handler/src/main/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueue.java b/handler/src/main/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueue.java
--- a/handler/src/main/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueue.java
+++ b/handler/src/main/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueue.java
@@ -42,38 +42,27 @@ abstract class SslHandlerCoalescingBufferQueue extends AbstractCoalescingBufferQ
 
     @Override
     protected ByteBuf compose(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf next) {
-        final int wrapDataSize = wrapDataSize();
-        if (cumulation instanceof CompositeByteBuf) {
-            CompositeByteBuf composite = (CompositeByteBuf) cumulation;
-            int numComponents = composite.numComponents();
-            if (numComponents == 0 ||
-                    !attemptCopyToCumulation(composite.internalComponent(numComponents - 1), next, wrapDataSize)) {
-                composite.addComponent(true, next);
-            }
-            return composite;
-        }
-        return attemptCopyToCumulation(cumulation, next, wrapDataSize) ? cumulation :
+        return attemptCopyToCumulation(cumulation, next, wrapDataSize()) ? cumulation :
                 copyAndCompose(alloc, cumulation, next);
     }
 
     @Override
-    protected ByteBuf composeFirst(ByteBufAllocator allocator, ByteBuf first) {
-        if (first instanceof CompositeByteBuf) {
-            CompositeByteBuf composite = (CompositeByteBuf) first;
-            if (wantsDirectBuffer) {
-                first = allocator.directBuffer(composite.readableBytes());
-            } else {
-                first = allocator.heapBuffer(composite.readableBytes());
-            }
-            try {
-                first.writeBytes(composite);
-            } catch (Throwable cause) {
-                first.release();
-                PlatformDependent.throwException(cause);
-            }
-            composite.release();
+    protected ByteBuf composeFirst(ByteBufAllocator allocator, ByteBuf first, int bufferSize) {
+        final ByteBuf newFirst;
+        if (wantsDirectBuffer) {
+            newFirst = allocator.directBuffer(bufferSize);
+        } else {
+            newFirst = allocator.heapBuffer(bufferSize);
+        }
+        try {
+            newFirst.writeBytes(first);
+        } catch (Throwable cause) {
+            newFirst.release();
+            PlatformDependent.throwException(cause);
         }
-        return first;
+        assert !first.isReadable();
+        first.release();
+        return newFirst;
     }
 
     @Override
diff --git a/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java b/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java
--- a/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java
+++ b/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java
@@ -181,10 +181,10 @@ public final ByteBuf remove(ByteBufAllocator alloc, int bytes, ChannelPromise ag
 
                     bytes -= bufferBytes;
                     if (toReturn == null) {
-                        // if there are no more bytes in the queue after this, there's no reason to compose
-                        toReturn = bufferBytes == readableBytes
+                        // if there are no more bytes to read, there's no reason to compose
+                        toReturn = bytes == 0
                                 ? entryBuffer
-                                : composeFirst(alloc, entryBuffer);
+                                : composeFirst(alloc, entryBuffer, bufferBytes + bytes);
                     } else {
                         toReturn = compose(alloc, toReturn, entryBuffer);
                     }
@@ -333,7 +333,21 @@ protected final ByteBuf copyAndCompose(ByteBufAllocator alloc, ByteBuf cumulatio
     /**
      * Calculate the first {@link ByteBuf} which will be used in subsequent calls to
      * {@link #compose(ByteBufAllocator, ByteBuf, ByteBuf)}.
+     * @param bufferSize the optimal size of the buffer needed for cumulation
+     * @return the first buffer
      */
+    protected ByteBuf composeFirst(ByteBufAllocator allocator, ByteBuf first, int bufferSize) {
+        return composeFirst(allocator, first);
+    }
+
+    /**
+     * Calculate the first {@link ByteBuf} which will be used in subsequent calls to
+     * {@link #compose(ByteBufAllocator, ByteBuf, ByteBuf)}.
+     * This method is deprecated and will be removed in the future. Implementing classes should
+     * override {@link #composeFirst(ByteBufAllocator, ByteBuf, int)} instead.
+     * @deprecated Use {AbstractCoalescingBufferQueue#composeFirst(ByteBufAllocator, ByteBuf, int)}
+     */
+    @Deprecated
     protected ByteBuf composeFirst(ByteBufAllocator allocator, ByteBuf first) {
         return first;
     }

diff --git a/handler/src/test/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueueTest.java b/handler/src/test/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueueTest.java
--- a/handler/src/test/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueueTest.java
+++ b/handler/src/test/java/io/netty/handler/ssl/SslHandlerCoalescingBufferQueueTest.java
@@ -21,19 +21,82 @@
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.embedded.EmbeddedChannel;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.EnumSource;
 
 import java.nio.ByteBuffer;
+import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class SslHandlerCoalescingBufferQueueTest {
+    static final Supplier<ByteBuf> BYTEBUF_SUPPLIER = new Supplier<ByteBuf>() {
+        @Override
+        public ByteBuf get() {
+            ByteBuf buf = Unpooled.directBuffer(128);
+            buf.writeLong(1);
+            buf.writeLong(2);
+            return buf;
+        }
+    };
+    static final Function<ByteBuf, ByteBuf> NO_WRAPPER = new Function<ByteBuf, ByteBuf>() {
+        @Override
+        public ByteBuf apply(ByteBuf byteBuf) {
+            return byteBuf;
+        }
+    };
+    static final Function<ByteBuf, ByteBuf> DUPLICATE_WRAPPER = new Function<ByteBuf, ByteBuf>() {
+        @Override
+        public ByteBuf apply(ByteBuf byteBuf) {
+            return byteBuf.duplicate();
+        }
+    };
+    static final Function<ByteBuf, ByteBuf> SLICE_WRAPPER = new Function<ByteBuf, ByteBuf>() {
+        @Override
+        public ByteBuf apply(ByteBuf byteBuf) {
+            return byteBuf.slice();
+        }
+    };
+    static ByteBuffer createNioBuffer() {
+        ByteBuffer nioBuffer = ByteBuffer.allocateDirect(128);
+        nioBuffer.putLong(1);
+        nioBuffer.putLong(2);
+        nioBuffer.position(nioBuffer.limit());
+        nioBuffer.flip();
+        return nioBuffer;
+    }
+
+    enum CumulationTestScenario {
+        BASIC_NIO_BUFFER(new Supplier() {
+            @Override
+            public ByteBuf get() {
+                return Unpooled.wrappedBuffer(createNioBuffer());
+            }
+        }, NO_WRAPPER),
+        READ_ONLY_AND_DUPLICATE_NIO_BUFFER(new Supplier() {
+            @Override
+            public ByteBuf get() {
+                return Unpooled.wrappedBuffer(createNioBuffer().asReadOnlyBuffer());
+            }
+        }, DUPLICATE_WRAPPER),
+        BASIC_DIRECT_BUFFER(BYTEBUF_SUPPLIER, NO_WRAPPER),
+        DUPLICATE_DIRECT_BUFFER(BYTEBUF_SUPPLIER, DUPLICATE_WRAPPER),
+        SLICED_DIRECT_BUFFER(BYTEBUF_SUPPLIER, SLICE_WRAPPER);
+
+        final Supplier<ByteBuf> bufferSupplier;
+        final Function<ByteBuf, ByteBuf> bufferWrapper;
+
+        CumulationTestScenario(Supplier<ByteBuf> bufferSupplier, Function<ByteBuf, ByteBuf> bufferWrapper) {
+            this.bufferSupplier = bufferSupplier;
+            this.bufferWrapper = bufferWrapper;
+        }
+    }
 
     @ParameterizedTest
-    @ValueSource(booleans = { true, false })
-    public void testCumulation(boolean readOnlyAndDuplicate) {
+    @EnumSource(CumulationTestScenario.class)
+    public void testCumulation(CumulationTestScenario testScenario) {
         EmbeddedChannel channel = new EmbeddedChannel();
         SslHandlerCoalescingBufferQueue queue = new SslHandlerCoalescingBufferQueue(channel, 16, false) {
             @Override
@@ -42,21 +105,12 @@ protected int wrapDataSize() {
             }
         };
 
-        ByteBuffer nioBuffer = ByteBuffer.allocateDirect(128);
-        nioBuffer.putLong(0);
-        nioBuffer.putLong(0);
-        nioBuffer.putLong(0);
-        nioBuffer.flip();
-        ByteBuf first;
-        if (readOnlyAndDuplicate) {
-            first = Unpooled.wrappedBuffer(nioBuffer.asReadOnlyBuffer()).duplicate();
-            first.writerIndex(8);
-        } else {
-            first = Unpooled.wrappedBuffer(nioBuffer);
-            first.writerIndex(8);
-        }
+        ByteBuf original = testScenario.bufferSupplier.get();
+        original.writerIndex(8);
+        ByteBuf first = testScenario.bufferWrapper.apply(original);
+        first.retain();
         queue.add(first);
-        ByteBuf second = Unpooled.copyLong(1);
+        ByteBuf second = Unpooled.copyLong(3);
         queue.add(second);
 
         ChannelPromise promise = channel.newPromise();
@@ -64,12 +118,16 @@ protected int wrapDataSize() {
         ByteBuf buffer = queue.remove(UnpooledByteBufAllocator.DEFAULT, 128, promise);
         try {
             assertEquals(16, buffer.readableBytes());
-            assertEquals(0, buffer.readLong());
             assertEquals(1, buffer.readLong());
+            assertEquals(3, buffer.readLong());
         } finally {
             buffer.release();
         }
         assertTrue(queue.isEmpty());
+        assertEquals(8, original.writerIndex());
+        original.writerIndex(original.capacity());
+        assertEquals(2, original.getLong(8));
+        first.release();
         assertEquals(0, first.refCnt());
         assertEquals(0, second.refCnt());
     }

SslHandler doesn't support a use case where the input ByteBuf is shared across multiple threads
### Expected behavior

There should be a way to pass shared ByteBuf instances to a ChannelPipeline that contains SslHandler. A ByteBuf instance might be cached and that's why it's shared. 

An alternative is to document this limitation clearly in SslHandler javadoc.

### Actual behavior

Currently this will cause problems such as https://github.com/apache/pulsar/issues/2401 reported in Pulsar in 2018.
This was mitigated by calling `.copy()` on the input ByteBuf instances with https://github.com/apache/pulsar/pull/2464 . However, this `.copy()` solution had another issue which is reported as #14068. That is mitigated in Pulsar with https://github.com/apache/pulsar/pull/22760 . 

### Steps to reproduce

It should be possible to reproduce this issue by concurrently passing a ByteBuf instance to multiple channel pipelines that contain a SslHandler.

### Minimal yet complete reproducer code (or URL to code)

System level repro case for Pulsar is documented in https://github.com/apache/pulsar/issues/22601#issuecomment-2122642708 . This is not minimal.

### Netty version

4.1.108

### JVM version (e.g. `java -version`)

openjdk version "17.0.11" 2024-04-16
OpenJDK Runtime Environment Temurin-17.0.11+9 (build 17.0.11+9)
OpenJDK 64-Bit Server VM Temurin-17.0.11+9 (build 17.0.11+9, mixed mode, sharing)

### OS version (e.g. `uname -a`)

Linux pop-os 6.8.0-76060800daily20240311-generic #202403110203171407766522.04~4c8e9a0 SMP PREEMPT_DYNAMIC Thu A x86_64 x86_64 x86_64 GNU/Linux

Passing the same `ByteBuf` instance to different `ChannelPipeline` is not supported at all. This is not really a limitation of the `SslHandler` but just the fact that `ByteBuf` is not thread-safe at all and once you pass a `ByteBuf` its ownership transfers.
> Passing the same `ByteBuf` instance to different `ChannelPipeline` is not supported at all. This is not really a limitation of the `SslHandler` but just the fact that `ByteBuf` is not thread-safe at all and once you pass a `ByteBuf` its ownership transfers.

Understood. I guess the main problem is that it's hard to find where this is documented. 
Many Netty users might have the assumption that you could use `.retainedDuplicate()` in cases where ownership is transfered. Is that the recommended approach for ownership transfer in a use case where a cached ByteBuf is sent to multiple clients?
In Pulsar, the `.copy()` solution was introduced in 2018 with apache/pulsar#2464 to fix another problem (apache/pulsar#2401) with Netty 4.1.22 which was used at that time.

There doesn't seem to be anything in SslHandler that would be causing the problems so I think that this issue that I have now reported here isn't a valid issue.

Any recommendations for sufficient ownership transfer to SslHandler will be appreciated since I'd like to make changes to Apache Pulsar (apache/pulsar#22760) based on Netty maintainer's recommendation. This issue can be closed. (I'll leave this open until someone replies. Thanks!)
I've been trying to find out the answer to my own question. It looks like a read-only buffer is needed with SslHandler for the use case where a cached ByteBuf is sent to multiple clients. It would be useful to document this in the SslHandler javadoc. 
If the input ByteBuf isn't read-only it looks like SslHandlerCoalescingBufferQueue would attempt to concatenate the buffers when possible in [attemptCopyToCumulation](https://github.com/netty/netty/blob/32a025b4eb2c9fefb389ecd5c781f7cabc6ab139/handler/src/main/java/io/netty/handler/ssl/SslHandler.java#L2436-L2451).
When I pass a read-only buffer, it logs a warning. I can file this as a separate issue if that makes it easier to track. It seems that the attemptCopyToCumulation method should check if the buffer is readonly before attempting to write to it.

```
2024-05-22T18:11:41,678+0000 [pulsar-io-3-4] WARN  org.apache.pulsar.broker.service.ServerCnx - [/10.34.1.241:60688] Got exception java.nio.ReadOnlyBufferException
   at io.netty.buffer.ReadOnlyByteBufferBuf.setBytes(ReadOnlyByteBufferBuf.java:384)
   at io.netty.buffer.DuplicatedByteBuf.setBytes(DuplicatedByteBuf.java:343)
   at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1104)
   at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1096)
   at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1087)
   at io.netty.handler.ssl.SslHandler.attemptCopyToCumulation(SslHandler.java:2446)
   at io.netty.handler.ssl.SslHandler.access$2800(SslHandler.java:171)
   at io.netty.handler.ssl.SslHandler$SslHandlerCoalescingBufferQueue.compose(SslHandler.java:2406)
   at io.netty.channel.AbstractCoalescingBufferQueue.remove(AbstractCoalescingBufferQueue.java:189)
   at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:841)
   at io.netty.handler.ssl.SslHandler.wrapAndFlush(SslHandler.java:822)
   at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:803)
   at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:941)
   at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:921)
   at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:907)
   at io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelOutboundHandlerAdapter.java:125)
   at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:939)
   at io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:957)
   at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:982)
   at io.netty.channel.AbstractChannelHandlerContext.writeAndFlush(AbstractChannelHandlerContext.java:950)
   at org.apache.pulsar.broker.service.PulsarCommandSenderImpl.lambda$sendMessagesToConsumer$1(PulsarCommandSenderImpl.java:286)
   at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
   at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
   at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
   at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:413)
   at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
   at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
   at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
   at java.base/java.lang.Thread.run(Thread.java:840)
```

> When I pass a read-only buffer, it logs a warning. I can file this as a separate issue if that makes it easier to track. It seems that the attemptCopyToCumulation method should check if the buffer is readonly before attempting to write to it.
> 
> ```
> 2024-05-22T18:11:41,678+0000 [pulsar-io-3-4] WARN  org.apache.pulsar.broker.service.ServerCnx - [/10.34.1.241:60688] Got exception java.nio.ReadOnlyBufferException
>    at io.netty.buffer.ReadOnlyByteBufferBuf.setBytes(ReadOnlyByteBufferBuf.java:384)
>    at io.netty.buffer.DuplicatedByteBuf.setBytes(DuplicatedByteBuf.java:343)
>    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1104)
>    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1096)
>    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1087)
>    at io.netty.handler.ssl.SslHandler.attemptCopyToCumulation(SslHandler.java:2446)
>    at io.netty.handler.ssl.SslHandler.access$2800(SslHandler.java:171)
>    at io.netty.handler.ssl.SslHandler$SslHandlerCoalescingBufferQueue.compose(SslHandler.java:2406)
>    at io.netty.channel.AbstractCoalescingBufferQueue.remove(AbstractCoalescingBufferQueue.java:189)
>    at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:841)
>    at io.netty.handler.ssl.SslHandler.wrapAndFlush(SslHandler.java:822)
>    at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:803)
>    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:941)
>    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:921)
>    at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:907)
>    at io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelOutboundHandlerAdapter.java:125)
>    at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:939)
>    at io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:957)
>    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:982)
>    at io.netty.channel.AbstractChannelHandlerContext.writeAndFlush(AbstractChannelHandlerContext.java:950)
>    at org.apache.pulsar.broker.service.PulsarCommandSenderImpl.lambda$sendMessagesToConsumer$1(PulsarCommandSenderImpl.java:286)
>    at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
>    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
>    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
>    at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:413)
>    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
>    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
>    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
>    at java.base/java.lang.Thread.run(Thread.java:840)
> ```

Yes please! Or do a PR :) 
> In Pulsar, the `.copy()` solution was introduced in 2018 with [apache/pulsar#2464](https://github.com/apache/pulsar/pull/2464) to fix another problem ([apache/pulsar#2401](https://github.com/apache/pulsar/issues/2401)) with Netty 4.1.22 which was used at that time.
> 
> There doesn't seem to be anything in SslHandler that would be causing the problems so I think that this issue that I have now reported here isn't a valid issue.
> 
> Any recommendations for sufficient ownership transfer to SslHandler will be appreciated since I'd like to make changes to Apache Pulsar ([apache/pulsar#22760](https://github.com/apache/pulsar/pull/22760)) based on Netty maintainer's recommendation. This issue can be closed. (I'll leave this open until someone replies. Thanks!)

I will review in detail and come back to you 
> Yes please! Or do a PR :)

@normanmaurer Here's the draft PR: #14071 . No unit tests yet.
Note that the JDK SSLEngine has had issues with read-only buffers in the past, because it likes to do encryption/decryption in-place. I don't think this is a well-trodden path.
> Note that the JDK SSLEngine has had issues with read-only buffers in the past, because it likes to do encryption/decryption in-place. I don't think this is a well-trodden path.

@chrisvest Isn't that already handled with the `MERGE_CUMULATOR`? https://github.com/netty/netty/blob/7d51e908e6146fba59269d567618949ee1b11899/handler/src/main/java/io/netty/handler/ssl/SslHandler.java#L304
https://github.com/netty/netty/blob/a09cc1620c9f8ac0cb17851b85a051459ee02964/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L98

If not, I guess it should be possible to add more changes for handling read-only buffers in [`SslHandlerCoalescingBufferQueue`](https://github.com/netty/netty/blob/7d51e908e6146fba59269d567618949ee1b11899/handler/src/main/java/io/netty/handler/ssl/SslHandler.java#L2388-L2428)?






@normanmaurer there's another obstacle in supporting this use case where where a cached ByteBuf is sent to multiple clients use SslHandler. I believe that this is a very common and generic use case and it exists in Apache Pulsar. 
The solution idea was to first call `.asReadOnly()` for the original source ByteBuf before adding it to the cache.
However, this won't work since the read-only buffer doesn't support reference counting in the same way as a `.duplicate()` buffer. I've created another issue #14075 about this.
@chrisvest Thanks for mentioning the JDK SSLEngine read-only buffer issue. I now found your email https://mail.openjdk.org/pipermail/security-dev/2022-March/029478.html 
and the related JDK bug reports https://bugs.openjdk.org/browse/JDK-8283577 and https://bugs.openjdk.org/browse/JDK-8285603. It seems that this is fixed [since Java jdk-19+23](https://github.com/openjdk/jdk/commit/f17c68ce4a0b4f5c3131f4e4626a5a55b7f2f61f).
@chrisvest @normanmaurer I have created a PR #14079 to address the JDK SSLEngine read-only buffer issue. It seems that after that is included, there would be support for the use case that this issue was about, supporting a use case where a cached ByteBuf is sent to multiple clients using SslHandler. I'd like to start using this solution in Apache Pulsar to replace the former solution where we always make a copy of all input when TLS is enabled. 
The idea will be to make the cached buffer read-only and that should be the way to signal SslHandler that it should make a copy of the input buffer if the SslEngineType wants a mutable buffer.
These are the PRs that are needed to support the use case: #14072, #14071, #14076 and #14078 (already merged to 4.1 branch). (PR #14079 isn't needed after all).
org.io.netty.handler.ssl.SslHandlerCoalescingBufferQueueTest