diff --git a/buffer/src/main/java/io/netty/buffer/api/DefaultGlobalBufferAllocator.java b/buffer/src/main/java/io/netty/buffer/api/DefaultGlobalBufferAllocator.java
new file mode 100644
index 0000000000..303522223c
--- /dev/null
+++ b/buffer/src/main/java/io/netty/buffer/api/DefaultGlobalBufferAllocator.java
@@ -0,0 +1,83 @@
+/*
+ * 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.util.internal.PlatformDependent;
+import io.netty.util.internal.SystemPropertyUtil;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+
+import java.util.Locale;
+import java.util.function.Supplier;
+
+import static io.netty.buffer.api.BufferAllocator.offHeapPooled;
+import static io.netty.buffer.api.BufferAllocator.offHeapUnpooled;
+import static io.netty.buffer.api.BufferAllocator.onHeapPooled;
+import static io.netty.buffer.api.BufferAllocator.onHeapUnpooled;
+import static io.netty.util.internal.ObjectUtil.checkNotNullWithIAE;
+import static io.netty.util.internal.PlatformDependent.directBufferPreferred;
+import static java.lang.Runtime.getRuntime;
+
+/**
+ * A {@link BufferAllocator} which is {@link #close() disposed} when the {@link Runtime} is shutdown.
+ */
+public final class DefaultGlobalBufferAllocator implements BufferAllocator {
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultGlobalBufferAllocator.class);
+    public static final BufferAllocator DEFAUL_GLOBAL_BUFFER_ALLOCATOR;
+
+    static {
+        String allocType = SystemPropertyUtil.get(
+                "io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
+        allocType = allocType.toLowerCase(Locale.US).trim();
+
+        BufferAllocator alloc;
+        if ("unpooled".equals(allocType)) {
+            alloc = directBufferPreferred() ? offHeapUnpooled() : onHeapUnpooled();
+            logger.debug("-Dio.netty.allocator.type: {}", allocType);
+        } else if ("pooled".equals(allocType)) {
+            alloc = directBufferPreferred() ? offHeapPooled() : onHeapPooled();
+            logger.debug("-Dio.netty.allocator.type: {}", allocType);
+        } else {
+            alloc = directBufferPreferred() ? offHeapPooled() : onHeapPooled();
+            logger.debug("-Dio.netty.allocator.type: pooled (unknown: {})", allocType);
+        }
+        DEFAUL_GLOBAL_BUFFER_ALLOCATOR = new DefaultGlobalBufferAllocator(alloc);
+    }
+
+    private final BufferAllocator delegate;
+
+    private DefaultGlobalBufferAllocator(BufferAllocator delegate) {
+        this.delegate = checkNotNullWithIAE(delegate, "delegate");
+        getRuntime().addShutdownHook(new Thread(this.delegate::close));
+    }
+
+    @Override
+    public Buffer allocate(int size) {
+        return delegate.allocate(size);
+    }
+
+    @Override
+    public Supplier<Buffer> constBufferSupplier(byte[] bytes) {
+        return delegate.constBufferSupplier(bytes);
+    }
+
+    /**
+     * @throws UnsupportedOperationException Close is not supported as this is a shared allocator.
+     */
+    @Override
+    public void close() {
+        throw new UnsupportedOperationException("Global buffer allocator can not be closed explicitly.");
+    }
+}
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/DelegatingChannelHandlerContext.java b/codec-http/src/main/java/io/netty/handler/codec/http/DelegatingChannelHandlerContext.java
index 2236766155..bd5d85c2ff 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/DelegatingChannelHandlerContext.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/DelegatingChannelHandlerContext.java
@@ -16,6 +16,7 @@
 package io.netty.handler.codec.http;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -139,6 +140,11 @@ abstract class DelegatingChannelHandlerContext implements ChannelHandlerContext
         return ctx.alloc();
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return ctx.bufferAllocator();
+    }
+
     @Deprecated
     public <T> Attribute<T> attr(AttributeKey<T> key) {
         return ctx.attr(key);
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameInboundWriter.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameInboundWriter.java
index 090a9fd628..2b5db4fa9a 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameInboundWriter.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameInboundWriter.java
@@ -18,6 +18,7 @@ package io.netty.handler.codec.http2;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -213,6 +214,11 @@ final class Http2FrameInboundWriter {
             return channel.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return channel.bufferAllocator();
+        }
+
         @Override
         public <T> Attribute<T> attr(AttributeKey<T> key) {
             return channel.attr(key);
diff --git a/codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java b/codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java
index 90f09ad069..7c9d82b8b3 100644
--- a/codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java
+++ b/codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java
@@ -17,6 +17,7 @@ package io.netty.handler.codec.memcache.binary;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -207,6 +208,11 @@ public final class BinaryMemcacheClientCodec extends
                     return ctx.alloc();
                 }
 
+                @Override
+                public BufferAllocator bufferAllocator() {
+                    return ctx.bufferAllocator();
+                }
+
                 @Override
                 @Deprecated
                 public <T> Attribute<T> attr(AttributeKey<T> key) {
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
index 6e386cf481..eb37e0c36b 100644
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -20,6 +20,7 @@ import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufConvertible;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelHandler;
@@ -626,6 +627,11 @@ public abstract class ByteToMessageDecoder extends ChannelHandlerAdapter {
             return ctx.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return ctx.bufferAllocator();
+        }
+
         @Override
         @Deprecated
         public <T> Attribute<T> attr(AttributeKey<T> key) {
diff --git a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelHandlerContext.java b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelHandlerContext.java
index ad05e775dc..4c8a99f3fc 100644
--- a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelHandlerContext.java
+++ b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelHandlerContext.java
@@ -15,6 +15,7 @@
 package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -37,11 +38,22 @@ public abstract class EmbeddedChannelHandlerContext implements ChannelHandlerCon
     private final EventLoop eventLoop;
     private final Channel channel;
     private final ByteBufAllocator alloc;
+    private final BufferAllocator bufferAllocator;
     private final ChannelHandler handler;
     private SocketAddress localAddress;
 
     protected EmbeddedChannelHandlerContext(ByteBufAllocator alloc, ChannelHandler handler, EmbeddedChannel channel) {
         this.alloc = requireNonNull(alloc, "alloc");
+        this.bufferAllocator = null;
+        this.channel = requireNonNull(channel, "channel");
+        this.handler = requireNonNull(handler, "handler");
+        eventLoop = requireNonNull(channel.executor(), "eventLoop");
+    }
+
+    protected EmbeddedChannelHandlerContext(BufferAllocator bufferAllocator, ChannelHandler handler,
+                                            EmbeddedChannel channel) {
+        this.bufferAllocator = requireNonNull(bufferAllocator, "bufferAllocator");
+        this.alloc = null;
         this.channel = requireNonNull(channel, "channel");
         this.handler = requireNonNull(handler, "handler");
         eventLoop = requireNonNull(channel.executor(), "eventLoop");
@@ -244,6 +256,11 @@ public abstract class EmbeddedChannelHandlerContext implements ChannelHandlerCon
         return alloc;
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return bufferAllocator;
+    }
+
     @Override
     public final Promise<Void> newPromise() {
         return channel().newPromise();
diff --git a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteAccumulatingHandlerContext.java b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteAccumulatingHandlerContext.java
index a1d9d77cae..e59afd02db 100644
--- a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteAccumulatingHandlerContext.java
+++ b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteAccumulatingHandlerContext.java
@@ -16,6 +16,7 @@ package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.ByteToMessageDecoder;
@@ -28,13 +29,25 @@ public abstract class EmbeddedChannelWriteAccumulatingHandlerContext extends Emb
     private final ByteToMessageDecoder.Cumulator cumulator;
 
     protected EmbeddedChannelWriteAccumulatingHandlerContext(ByteBufAllocator alloc, ChannelHandler handler,
-                                                          ByteToMessageDecoder.Cumulator writeCumulator) {
+                                                             ByteToMessageDecoder.Cumulator writeCumulator) {
         this(alloc, handler, writeCumulator, new EmbeddedChannel());
     }
 
     protected EmbeddedChannelWriteAccumulatingHandlerContext(ByteBufAllocator alloc, ChannelHandler handler,
-                                                          ByteToMessageDecoder.Cumulator writeCumulator,
-                                                          EmbeddedChannel channel) {
+                                                             ByteToMessageDecoder.Cumulator writeCumulator,
+                                                             EmbeddedChannel channel) {
+        super(alloc, handler, channel);
+        cumulator = requireNonNull(writeCumulator, "writeCumulator");
+    }
+
+    protected EmbeddedChannelWriteAccumulatingHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                                             ByteToMessageDecoder.Cumulator writeCumulator) {
+        this(alloc, handler, writeCumulator, new EmbeddedChannel());
+    }
+
+    protected EmbeddedChannelWriteAccumulatingHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                                             ByteToMessageDecoder.Cumulator writeCumulator,
+                                                             EmbeddedChannel channel) {
         super(alloc, handler, channel);
         cumulator = requireNonNull(writeCumulator, "writeCumulator");
     }
diff --git a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteReleaseHandlerContext.java b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteReleaseHandlerContext.java
index 66135daef8..2b79bee3f9 100644
--- a/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteReleaseHandlerContext.java
+++ b/microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteReleaseHandlerContext.java
@@ -15,6 +15,7 @@
 package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.util.ReferenceCounted;
@@ -29,6 +30,14 @@ public abstract class EmbeddedChannelWriteReleaseHandlerContext extends Embedded
             EmbeddedChannel channel) {
         super(alloc, handler, channel);
     }
+    protected EmbeddedChannelWriteReleaseHandlerContext(BufferAllocator alloc, ChannelHandler handler) {
+        this(alloc, handler, new EmbeddedChannel());
+    }
+
+    protected EmbeddedChannelWriteReleaseHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+            EmbeddedChannel channel) {
+        super(alloc, handler, channel);
+    }
 
     @Override
     protected abstract void handleException(Throwable t);
diff --git a/microbench/src/main/java/io/netty/microbench/handler/ssl/AbstractSslHandlerBenchmark.java b/microbench/src/main/java/io/netty/microbench/handler/ssl/AbstractSslHandlerBenchmark.java
index 974899831c..bb11e2c7a3 100644
--- a/microbench/src/main/java/io/netty/microbench/handler/ssl/AbstractSslHandlerBenchmark.java
+++ b/microbench/src/main/java/io/netty/microbench/handler/ssl/AbstractSslHandlerBenchmark.java
@@ -17,6 +17,7 @@ package io.netty.microbench.handler.ssl;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.ssl.SslContext;
@@ -164,7 +165,12 @@ public class AbstractSslHandlerBenchmark extends AbstractMicrobenchmark {
     private static final class SslThroughputBenchmarkHandlerContext extends
             EmbeddedChannelWriteAccumulatingHandlerContext {
         SslThroughputBenchmarkHandlerContext(ByteBufAllocator alloc, ChannelHandler handler,
-                                                    ByteToMessageDecoder.Cumulator writeCumulator) {
+                                             ByteToMessageDecoder.Cumulator writeCumulator) {
+            super(alloc, handler, writeCumulator);
+        }
+
+        SslThroughputBenchmarkHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                             ByteToMessageDecoder.Cumulator writeCumulator) {
             super(alloc, handler, writeCumulator);
         }
 
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelConfig.java
index 280b5c499f..3286581fa0 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelConfig.java
@@ -17,6 +17,8 @@ package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
 
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -56,6 +58,12 @@ public class EpollChannelConfig extends DefaultChannelConfig {
         return this;
     }
 
+    @Override
+    public EpollChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         if (!(allocator.newHandle() instanceof RecvByteBufAllocator.ExtendedHandle)) {
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
index 45ac63d697..2b9a54d0f1 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java
@@ -17,6 +17,7 @@ package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -220,6 +221,12 @@ public final class EpollDatagramChannelConfig extends EpollChannelConfig impleme
         return this;
     }
 
+    @Override
+    public EpollDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDatagramChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
         super.setConnectTimeoutMillis(connectTimeoutMillis);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainDatagramChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainDatagramChannelConfig.java
index fd31542070..07b17c9f82 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainDatagramChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainDatagramChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -96,6 +97,12 @@ public final class EpollDomainDatagramChannelConfig extends EpollChannelConfig i
         return this;
     }
 
+    @Override
+    public EpollDomainDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDomainDatagramChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java
index a6a04a9d53..820a557cfd 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java
@@ -18,6 +18,7 @@ package io.netty.channel.epoll;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -113,6 +114,12 @@ public final class EpollDomainSocketChannelConfig extends EpollDuplexChannelConf
         return this;
     }
 
+    @Override
+    public EpollDomainSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDomainSocketChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDuplexChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDuplexChannelConfig.java
index e522d757cd..5501cf35be 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDuplexChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDuplexChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -94,6 +95,12 @@ public class EpollDuplexChannelConfig extends EpollChannelConfig implements Dupl
         return this;
     }
 
+    @Override
+    public EpollDuplexChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDuplexChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerChannelConfig.java
index 43ba1f0262..7d8da47f8d 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -187,6 +188,12 @@ public class EpollServerChannelConfig extends EpollChannelConfig implements Serv
         return this;
     }
 
+    @Override
+    public EpollServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerSocketChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerSocketChannelConfig.java
index 014604e1db..4af48b9967 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerSocketChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerSocketChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -134,6 +135,12 @@ public final class EpollServerSocketChannelConfig extends EpollServerChannelConf
         return this;
     }
 
+    @Override
+    public EpollServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannelConfig.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannelConfig.java
index 37f6d12f30..c4949a5d81 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannelConfig.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -593,6 +594,12 @@ public final class EpollSocketChannelConfig extends EpollDuplexChannelConfig imp
         return this;
     }
 
+    @Override
+    public EpollSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueChannelConfig.java
index c144ffd555..2b2d008ada 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueChannelConfig.java
@@ -16,6 +16,8 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
@@ -108,6 +110,12 @@ public class KQueueChannelConfig extends DefaultChannelConfig {
         return this;
     }
 
+    @Override
+    public KQueueChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         if (!(allocator.newHandle() instanceof RecvByteBufAllocator.ExtendedHandle)) {
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannelConfig.java
index ee6b070e66..ad5860be06 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -235,6 +236,12 @@ public final class KQueueDatagramChannelConfig extends KQueueChannelConfig imple
         return this;
     }
 
+    @Override
+    public KQueueDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDatagramChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
         super.setConnectTimeoutMillis(connectTimeoutMillis);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainDatagramChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainDatagramChannelConfig.java
index 55d73e92f7..a30d9a18be 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainDatagramChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainDatagramChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -96,6 +97,12 @@ public final class KQueueDomainDatagramChannelConfig
         return this;
     }
 
+    @Override
+    public KQueueDomainDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDomainDatagramChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java
index 219528997c..519f58d277 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -115,6 +116,12 @@ public final class KQueueDomainSocketChannelConfig extends KQueueDuplexChannelCo
         return this;
     }
 
+    @Override
+    public KQueueDomainSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDomainSocketChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDuplexChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDuplexChannelConfig.java
index 0f50ae4d64..38d3f7202e 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDuplexChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDuplexChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -96,6 +97,12 @@ public class KQueueDuplexChannelConfig extends KQueueChannelConfig implements Du
         return this;
     }
 
+    @Override
+    public KQueueDuplexChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDuplexChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerChannelConfig.java
index 81e3777625..73d0f2151f 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -164,6 +165,12 @@ public class KQueueServerChannelConfig extends KQueueChannelConfig implements Se
         return this;
     }
 
+    @Override
+    public KQueueServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerSocketChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerSocketChannelConfig.java
index 3bbe935548..b7a7f9cf5b 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerSocketChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerSocketChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -161,6 +162,12 @@ public class KQueueServerSocketChannelConfig extends KQueueServerChannelConfig i
         return this;
     }
 
+    @Override
+    public KQueueServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueSocketChannelConfig.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueSocketChannelConfig.java
index 1ad15083c0..61fd377f36 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueSocketChannelConfig.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueSocketChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -349,6 +350,12 @@ public final class KQueueSocketChannelConfig extends KQueueDuplexChannelConfig i
         return this;
     }
 
+    @Override
+    public KQueueSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java b/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java
index 8616407624..ceee3a8525 100644
--- a/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java
+++ b/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java
@@ -18,6 +18,8 @@ package io.netty.channel.sctp;
 import com.sun.nio.sctp.SctpChannel;
 import com.sun.nio.sctp.SctpStandardSocketOptions;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -202,6 +204,12 @@ public class DefaultSctpChannelConfig extends DefaultChannelConfig implements Sc
         return this;
     }
 
+    @Override
+    public SctpChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SctpChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java b/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java
index 155df9030d..94c2d624af 100644
--- a/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java
+++ b/transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java
@@ -21,6 +21,8 @@ import static java.util.Objects.requireNonNull;
 import com.sun.nio.sctp.SctpServerChannel;
 import com.sun.nio.sctp.SctpStandardSocketOptions;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -183,6 +185,12 @@ public class DefaultSctpServerChannelConfig extends DefaultChannelConfig impleme
         return this;
     }
 
+    @Override
+    public SctpServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SctpServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport/src/main/java/io/netty/channel/Channel.java b/transport/src/main/java/io/netty/channel/Channel.java
index 47a6746386..90d14d5793 100644
--- a/transport/src/main/java/io/netty/channel/Channel.java
+++ b/transport/src/main/java/io/netty/channel/Channel.java
@@ -17,6 +17,7 @@ package io.netty.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.DatagramPacket;
 import io.netty.channel.socket.ServerSocketChannel;
@@ -199,11 +200,20 @@ public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparabl
 
     /**
      * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
+     * @deprecated Use {@link #bufferAllocator()}
      */
+    @Deprecated
     default ByteBufAllocator alloc() {
         return config().getAllocator();
     }
 
+    /**
+     * Return the assigned {@link BufferAllocator} which will be used to allocate {@link io.netty.buffer.api.Buffer}s.
+     */
+    default BufferAllocator bufferAllocator() {
+        return config().getBufferAllocator();
+    }
+
     @Override
     default Channel read() {
         pipeline().read();
diff --git a/transport/src/main/java/io/netty/channel/ChannelConfig.java b/transport/src/main/java/io/netty/channel/ChannelConfig.java
index d3d5b64833..043c020f4e 100644
--- a/transport/src/main/java/io/netty/channel/ChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/ChannelConfig.java
@@ -16,6 +16,7 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.socket.SocketChannelConfig;
 
 import java.nio.ByteBuffer;
@@ -165,15 +166,31 @@ public interface ChannelConfig {
     /**
      * Returns {@link ByteBufAllocator} which is used for the channel
      * to allocate buffers.
+     * @deprecated use {@link #getBufferAllocator()}
      */
+    @Deprecated
     ByteBufAllocator getAllocator();
 
     /**
      * Set the {@link ByteBufAllocator} which is used for the channel
      * to allocate buffers.
+     * @deprecated use {@link #setBufferAllocator(BufferAllocator)}
      */
+    @Deprecated
     ChannelConfig setAllocator(ByteBufAllocator allocator);
 
+    /**
+     * Returns {@link BufferAllocator} which is used for the channel
+     * to allocate buffers.
+     */
+    BufferAllocator getBufferAllocator();
+
+    /**
+     * Set the {@link ByteBufAllocator} which is used for the channel
+     * to allocate buffers.
+     */
+    ChannelConfig setBufferAllocator(BufferAllocator allocator);
+
     /**
      * Returns {@link RecvByteBufAllocator} which is used for the channel to allocate receive buffers.
      */
diff --git a/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java
index 2894ac9978..5c21abceed 100644
--- a/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java
+++ b/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java
@@ -17,6 +17,7 @@ package io.netty.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.AttributeMap;
@@ -186,8 +187,14 @@ public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvok
     /**
      * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
      */
+    @Deprecated
     ByteBufAllocator alloc();
 
+    /**
+     * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
+     */
+    BufferAllocator bufferAllocator();
+
     /**
      * @deprecated Use {@link Channel#attr(AttributeKey)}
      */
diff --git a/transport/src/main/java/io/netty/channel/ChannelOption.java b/transport/src/main/java/io/netty/channel/ChannelOption.java
index ac4e1f56c7..4bded6f281 100644
--- a/transport/src/main/java/io/netty/channel/ChannelOption.java
+++ b/transport/src/main/java/io/netty/channel/ChannelOption.java
@@ -18,6 +18,7 @@ package io.netty.channel;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.AbstractConstant;
 import io.netty.util.ConstantPool;
 
@@ -77,6 +78,7 @@ public class ChannelOption<T> extends AbstractConstant<ChannelOption<T>> {
     }
 
     public static final ChannelOption<ByteBufAllocator> ALLOCATOR = valueOf("ALLOCATOR");
+    public static final ChannelOption<BufferAllocator> BUFFER_ALLOCATOR = valueOf("BUFFER_ALLOCATOR");
     public static final ChannelOption<RecvByteBufAllocator> RCVBUF_ALLOCATOR = valueOf("RCVBUF_ALLOCATOR");
     public static final ChannelOption<MessageSizeEstimator> MESSAGE_SIZE_ESTIMATOR = valueOf("MESSAGE_SIZE_ESTIMATOR");
 
diff --git a/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java b/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java
index bed0260de0..948ff490e3 100644
--- a/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java
+++ b/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java
@@ -16,6 +16,7 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.EventExecutor;
@@ -493,6 +494,11 @@ public class CombinedChannelDuplexHandler<I extends ChannelHandler, O extends Ch
             return ctx.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return ctx.bufferAllocator();
+        }
+
         @Override
         public Promise<Void> newPromise() {
             return ctx.newPromise();
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
index 28b6ae0a36..6ba204a57f 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
@@ -16,6 +16,8 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.buffer.api.DefaultGlobalBufferAllocator;
 import io.netty.util.internal.ObjectUtil;
 
 import java.util.IdentityHashMap;
@@ -27,6 +29,7 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import static io.netty.channel.ChannelOption.ALLOCATOR;
 import static io.netty.channel.ChannelOption.AUTO_CLOSE;
 import static io.netty.channel.ChannelOption.AUTO_READ;
+import static io.netty.channel.ChannelOption.BUFFER_ALLOCATOR;
 import static io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS;
 import static io.netty.channel.ChannelOption.MAX_MESSAGES_PER_READ;
 import static io.netty.channel.ChannelOption.MAX_MESSAGES_PER_WRITE;
@@ -57,6 +60,7 @@ public class DefaultChannelConfig implements ChannelConfig {
     protected final Channel channel;
 
     private volatile ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
+    private volatile BufferAllocator bufferAllocator = DefaultGlobalBufferAllocator.DEFAUL_GLOBAL_BUFFER_ALLOCATOR;
     private volatile RecvByteBufAllocator rcvBufAllocator;
     private volatile MessageSizeEstimator msgSizeEstimator = DEFAULT_MSG_SIZE_ESTIMATOR;
 
@@ -85,7 +89,7 @@ public class DefaultChannelConfig implements ChannelConfig {
         return getOptions(
                 null,
                 CONNECT_TIMEOUT_MILLIS, MAX_MESSAGES_PER_READ, WRITE_SPIN_COUNT,
-                ALLOCATOR, AUTO_READ, AUTO_CLOSE, RCVBUF_ALLOCATOR, WRITE_BUFFER_HIGH_WATER_MARK,
+                ALLOCATOR, BUFFER_ALLOCATOR, AUTO_READ, AUTO_CLOSE, RCVBUF_ALLOCATOR, WRITE_BUFFER_HIGH_WATER_MARK,
                 WRITE_BUFFER_LOW_WATER_MARK, WRITE_BUFFER_WATER_MARK, MESSAGE_SIZE_ESTIMATOR, MAX_MESSAGES_PER_WRITE);
     }
 
@@ -132,6 +136,9 @@ public class DefaultChannelConfig implements ChannelConfig {
         if (option == ALLOCATOR) {
             return (T) getAllocator();
         }
+        if (option == BUFFER_ALLOCATOR) {
+            return (T) getBufferAllocator();
+        }
         if (option == RCVBUF_ALLOCATOR) {
             return (T) getRecvByteBufAllocator();
         }
@@ -172,6 +179,8 @@ public class DefaultChannelConfig implements ChannelConfig {
             setWriteSpinCount((Integer) value);
         } else if (option == ALLOCATOR) {
             setAllocator((ByteBufAllocator) value);
+        } else if (option == BUFFER_ALLOCATOR) {
+            setBufferAllocator((BufferAllocator) value);
         } else if (option == RCVBUF_ALLOCATOR) {
             setRecvByteBufAllocator((RecvByteBufAllocator) value);
         } else if (option == AUTO_READ) {
@@ -297,6 +306,18 @@ public class DefaultChannelConfig implements ChannelConfig {
         return this;
     }
 
+    @Override
+    public BufferAllocator getBufferAllocator() {
+        return bufferAllocator;
+    }
+
+    @Override
+    public ChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        requireNonNull(bufferAllocator, "bufferAllocator");
+        this.bufferAllocator = bufferAllocator;
+        return this;
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     public <T extends RecvByteBufAllocator> T getRecvByteBufAllocator() {
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
index 26820703b8..8b30dc24ca 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
@@ -18,6 +18,7 @@ package io.netty.channel;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.ReferenceCountUtil;
@@ -131,6 +132,11 @@ final class DefaultChannelHandlerContext implements ChannelHandlerContext, Resou
         return channel().config().getAllocator();
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return channel().config().getBufferAllocator();
+    }
+
     @Override
     public String name() {
         return name;
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
index 33e4853d19..eee7360a93 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -386,6 +388,12 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
         return this;
     }
 
+    @Override
+    public DatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public DatagramChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
index 9f72e6a5a3..96168821d5 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java
@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -171,6 +173,12 @@ public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
         return this;
     }
 
+    @Override
+    public ServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public ServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);
diff --git a/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java b/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
index 1e05e5051c..65fc952211 100644
--- a/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java
@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -304,6 +306,12 @@ public class DefaultSocketChannelConfig extends DefaultChannelConfig
         return this;
     }
 
+    @Override
+    public SocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);