@normanmaurer @jasobrown - FYI. This should probably be reopened. My PR only fixes this for subclasses of ZlibDecoder; it does not fix this for any other compression algorithms.
Can you add to the doc describing what `0` does? same comment can this be private/final?    optional: check preferredSize is non-negative? +1 nit: `else` block not needed I think either the second `ensureWritable` param should be `false` or the return should be compared to `3` rather than `1`? For consistency shouldn't `maxAllocationReached` be called here in the `preferredSize < maxAllocation` case? Not sure that the semantics of the method are clear... if it's overridden to not throw then what state will the provided buffer be in, what is the impl expected to do with it and what is it expected to return? Maybe it would make most sense to pass false for the `force` param of `ensureWritable` above so that the buffer isn't changed, and also pass `preferredSize` and `ctx` to this method as additional parameters?

nit: and maybe clearer method name would be something like `decompressionBufferExhausted`? preferredSize was not checked by the existing code, and adding a check will actually duplicate the check because ByteBuf or ByteBufAllocator (whichever is ultimately called) will check that it's non-negative.

About the `preferredSize < maxAllocation` case, I'm not following you. preferredSize will often be less than maxAllocation since preferredSize is a guesstimate of how much more space is needed to finish decompressing the input. So e.g. maxAllocation might be set to 1 MiB but the decoder estimates it only needs 10 KiB to finish decompressing. I'm pretty sure this is correct as-is. The desired behavior is that if capacity < maxAllocation, we want to expand the buffer, even if we can't expand it by the preferredSize. Any expansion at all makes it possible that the decoder will successfully complete the decompression, since preferredSize is just a guesstimate. Hence we want to expand even if the expansion is less than what we'd prefer. If it turns out that's not enough space, then this method will be called again, and the buffer will already have capacity = maxAllocation and remaining = 0, and we'll fail at that point. You're right that this probably shouldn't return anything - I'll change that. Can't remember exactly what I was thinking when I did that. My use case for wanting to override this is so that I can potentially log the data that has been decompressed so far before throwing the exception. I'll make it final but leave it as protected in case a subclass wants to read it for any reason. @rdicroce apologies, I meant `preferredSize > maxAllocation` Since preferredSize is a guesstimate of how much space is needed, it's possible that the compression ratio is poor and we don't need that much space. So e.g. the compressed size could be 100 KiB, and so JdkZlibDecoder and ZlibDecoder estimate they need 200 KiB to decompress. So they pass preferredSize = 200 KiB but then it turns out the decompressed data is only 150 KiB in size. If maxAllocation is 175 KiB, then decompression ought to succeed since 150 < 175, even though 200 > 175. OK I follow now, I missed that subtlety of `ensureWritable`, thinking that `1` would only apply to the `force == false` case. This is the same confusion as the comment above and I see that your logic there is consistent. Sorry!

It probably wouldn't harm to add a comment noting that one "final" attempt will always still be made with the buffer at `maxAllocation` even if that is `< preferredSize`. nit: final  call `chDecoder.finish()` and assert the return value nit: protected (as this is an abstract class) nit: protected (as this is an abstract class) Good catch - finish() was throwing an exception because the input buffer wasn't being consumed and the decoders weren't marking themselves as finished, so another attempt was being made to read the buffer. I've updated JZlibDecoder and JdkZlibDecoder to consume the buffer and mark themselves as finished at the same time they release the decompression buffer.
Compression/Decompression Codecs should enforce memory allocation size limits
### Expected behavior
To protect against OOME the compression and decompression codecs should explicitly limit the amount of data they compress and decompress. We may be vulnerable to OOME from large or malicious input.

### Actual behavior
In light of https://github.com/netty/netty/pull/5997 most of the compression/decompression codecs don't enforce limits on buffer allocation sizes.

### Steps to reproduce
N/A

### Minimal yet complete reproducer code (or URL to code)
N/A

### Netty version
4.1.7-SNAPSHOT

### JVM version (e.g. `java -version`)
N/A

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

codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java
codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java
codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest1.java codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest2.java codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java
diff --git a/codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java b/codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java
index f87344ab599..6c65cd5fe6a 100644
--- a/codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/compression/JZlibDecoder.java
@@ -18,6 +18,7 @@
 import com.jcraft.jzlib.Inflater;
 import com.jcraft.jzlib.JZlib;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.internal.ObjectUtil;
 
@@ -35,7 +36,21 @@ public class JZlibDecoder extends ZlibDecoder {
      * @throws DecompressionException if failed to initialize zlib
      */
     public JZlibDecoder() {
-        this(ZlibWrapper.ZLIB);
+        this(ZlibWrapper.ZLIB, 0);
+    }
+
+    /**
+     * Creates a new instance with the default wrapper ({@link ZlibWrapper#ZLIB})
+     * and specified maximum buffer allocation.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     *
+     * @throws DecompressionException if failed to initialize zlib
+     */
+    public JZlibDecoder(int maxAllocation) {
+        this(ZlibWrapper.ZLIB, maxAllocation);
     }
 
     /**
@@ -44,6 +59,21 @@ public JZlibDecoder() {
      * @throws DecompressionException if failed to initialize zlib
      */
     public JZlibDecoder(ZlibWrapper wrapper) {
+        this(wrapper, 0);
+    }
+
+    /**
+     * Creates a new instance with the specified wrapper and maximum buffer allocation.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     *
+     * @throws DecompressionException if failed to initialize zlib
+     */
+    public JZlibDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        super(maxAllocation);
+
         ObjectUtil.checkNotNull(wrapper, "wrapper");
 
         int resultCode = z.init(ZlibUtil.convertWrapperType(wrapper));
@@ -60,6 +90,22 @@ public JZlibDecoder(ZlibWrapper wrapper) {
      * @throws DecompressionException if failed to initialize zlib
      */
     public JZlibDecoder(byte[] dictionary) {
+        this(dictionary, 0);
+    }
+
+    /**
+     * Creates a new instance with the specified preset dictionary and maximum buffer allocation.
+     * The wrapper is always {@link ZlibWrapper#ZLIB} because it is the only format that
+     * supports the preset dictionary.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     *
+     * @throws DecompressionException if failed to initialize zlib
+     */
+    public JZlibDecoder(byte[] dictionary, int maxAllocation) {
+        super(maxAllocation);
         this.dictionary = ObjectUtil.checkNotNull(dictionary, "dictionary");
         int resultCode;
         resultCode = z.inflateInit(JZlib.W_ZLIB);
@@ -105,11 +151,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
             final int oldNextInIndex = z.next_in_index;
 
             // Configure output.
-            ByteBuf decompressed = ctx.alloc().heapBuffer(inputLength << 1);
+            ByteBuf decompressed = prepareDecompressBuffer(ctx, null, inputLength << 1);
 
             try {
                 loop: for (;;) {
-                    decompressed.ensureWritable(z.avail_in << 1);
+                    decompressed = prepareDecompressBuffer(ctx, decompressed, z.avail_in << 1);
                     z.avail_out = decompressed.writableBytes();
                     z.next_out = decompressed.array();
                     z.next_out_index = decompressed.arrayOffset() + decompressed.writerIndex();
@@ -165,4 +211,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
             z.next_out = null;
         }
     }
+
+    @Override
+    protected void decompressionBufferExhausted(ByteBuf buffer) {
+        finished = true;
+    }
 }
diff --git a/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java b/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java
index 776d1e7816b..7e694222aaf 100644
--- a/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/compression/JdkZlibDecoder.java
@@ -16,6 +16,7 @@
 package io.netty.handler.codec.compression;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.internal.ObjectUtil;
 
@@ -65,7 +66,19 @@ private enum GzipState {
      * Creates a new instance with the default wrapper ({@link ZlibWrapper#ZLIB}).
      */
     public JdkZlibDecoder() {
-        this(ZlibWrapper.ZLIB, null, false);
+        this(ZlibWrapper.ZLIB, null, false, 0);
+    }
+
+    /**
+     * Creates a new instance with the default wrapper ({@link ZlibWrapper#ZLIB})
+     * and the specified maximum buffer allocation.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     */
+    public JdkZlibDecoder(int maxAllocation) {
+        this(ZlibWrapper.ZLIB, null, false, maxAllocation);
     }
 
     /**
@@ -74,7 +87,20 @@ public JdkZlibDecoder() {
      * supports the preset dictionary.
      */
     public JdkZlibDecoder(byte[] dictionary) {
-        this(ZlibWrapper.ZLIB, dictionary, false);
+        this(ZlibWrapper.ZLIB, dictionary, false, 0);
+    }
+
+    /**
+     * Creates a new instance with the specified preset dictionary and maximum buffer allocation.
+     * The wrapper is always {@link ZlibWrapper#ZLIB} because it is the only format that
+     * supports the preset dictionary.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     */
+    public JdkZlibDecoder(byte[] dictionary, int maxAllocation) {
+        this(ZlibWrapper.ZLIB, dictionary, false, maxAllocation);
     }
 
     /**
@@ -83,18 +109,41 @@ public JdkZlibDecoder(byte[] dictionary) {
      * supported atm.
      */
     public JdkZlibDecoder(ZlibWrapper wrapper) {
-        this(wrapper, null, false);
+        this(wrapper, null, false, 0);
+    }
+
+    /**
+     * Creates a new instance with the specified wrapper and maximum buffer allocation.
+     * Be aware that only {@link ZlibWrapper#GZIP}, {@link ZlibWrapper#ZLIB} and {@link ZlibWrapper#NONE} are
+     * supported atm.
+     *
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     */
+    public JdkZlibDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        this(wrapper, null, false, maxAllocation);
     }
 
     public JdkZlibDecoder(ZlibWrapper wrapper, boolean decompressConcatenated) {
-        this(wrapper, null, decompressConcatenated);
+        this(wrapper, null, decompressConcatenated, 0);
+    }
+
+    public JdkZlibDecoder(ZlibWrapper wrapper, boolean decompressConcatenated, int maxAllocation) {
+        this(wrapper, null, decompressConcatenated, maxAllocation);
     }
 
     public JdkZlibDecoder(boolean decompressConcatenated) {
-        this(ZlibWrapper.GZIP, null, decompressConcatenated);
+        this(ZlibWrapper.GZIP, null, decompressConcatenated, 0);
     }
 
-    private JdkZlibDecoder(ZlibWrapper wrapper, byte[] dictionary, boolean decompressConcatenated) {
+    public JdkZlibDecoder(boolean decompressConcatenated, int maxAllocation) {
+        this(ZlibWrapper.GZIP, null, decompressConcatenated, maxAllocation);
+    }
+
+    private JdkZlibDecoder(ZlibWrapper wrapper, byte[] dictionary, boolean decompressConcatenated, int maxAllocation) {
+        super(maxAllocation);
+
         ObjectUtil.checkNotNull(wrapper, "wrapper");
 
         this.decompressConcatenated = decompressConcatenated;
@@ -177,7 +226,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
             inflater.setInput(array);
         }
 
-        ByteBuf decompressed = ctx.alloc().heapBuffer(inflater.getRemaining() << 1);
+        ByteBuf decompressed = prepareDecompressBuffer(ctx, null, inflater.getRemaining() << 1);
         try {
             boolean readFooter = false;
             while (!inflater.needsInput()) {
@@ -208,7 +257,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
                     }
                     break;
                 } else {
-                    decompressed.ensureWritable(inflater.getRemaining() << 1);
+                    decompressed = prepareDecompressBuffer(ctx, decompressed, inflater.getRemaining() << 1);
                 }
             }
 
@@ -238,6 +287,11 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         }
     }
 
+    @Override
+    protected void decompressionBufferExhausted(ByteBuf buffer) {
+        finished = true;
+    }
+
     @Override
     protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
         super.handlerRemoved0(ctx);
diff --git a/codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java b/codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java
index d01bc6b4de7..26fd3e74a76 100644
--- a/codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/compression/ZlibDecoder.java
@@ -16,6 +16,8 @@
 package io.netty.handler.codec.compression;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
 
 /**
@@ -23,9 +25,72 @@
  */
 public abstract class ZlibDecoder extends ByteToMessageDecoder {
 
+    /**
+     * Maximum allowed size of the decompression buffer.
+     */
+    protected final int maxAllocation;
+
+    /**
+     * Same as {@link #ZlibDecoder(int)} with maxAllocation = 0.
+     */
+    public ZlibDecoder() {
+        this(0);
+    }
+
+    /**
+     * Construct a new ZlibDecoder.
+     * @param maxAllocation
+     *          Maximum size of the decompression buffer. Must be &gt;= 0.
+     *          If zero, maximum size is decided by the {@link ByteBufAllocator}.
+     */
+    public ZlibDecoder(int maxAllocation) {
+        if (maxAllocation < 0) {
+            throw new IllegalArgumentException("maxAllocation must be >= 0");
+        }
+        this.maxAllocation = maxAllocation;
+    }
+
     /**
      * Returns {@code true} if and only if the end of the compressed stream
      * has been reached.
      */
     public abstract boolean isClosed();
+
+    /**
+     * Allocate or expand the decompression buffer, without exceeding the maximum allocation.
+     * Calls {@link #decompressionBufferExhausted(ByteBuf)} if the buffer is full and cannot be expanded further.
+     */
+    protected ByteBuf prepareDecompressBuffer(ChannelHandlerContext ctx, ByteBuf buffer, int preferredSize) {
+        if (buffer == null) {
+            if (maxAllocation == 0) {
+                return ctx.alloc().heapBuffer(preferredSize);
+            }
+
+            return ctx.alloc().heapBuffer(Math.min(preferredSize, maxAllocation), maxAllocation);
+        }
+
+        // this always expands the buffer if possible, even if the expansion is less than preferredSize
+        // we throw the exception only if the buffer could not be expanded at all
+        // this means that one final attempt to deserialize will always be made with the buffer at maxAllocation
+        if (buffer.ensureWritable(preferredSize, true) == 1) {
+            // buffer must be consumed so subclasses don't add it to output
+            // we therefore duplicate it when calling decompressionBufferExhausted() to guarantee non-interference
+            // but wait until after to consume it so the subclass can tell how much output is really in the buffer
+            decompressionBufferExhausted(buffer.duplicate());
+            buffer.skipBytes(buffer.readableBytes());
+            throw new DecompressionException("Decompression buffer has reached maximum size: " + buffer.maxCapacity());
+        }
+
+        return buffer;
+    }
+
+    /**
+     * Called when the decompression buffer cannot be expanded further.
+     * Default implementation is a no-op, but subclasses can override in case they want to
+     * do something before the {@link DecompressionException} is thrown, such as log the
+     * data that was decompressed so far.
+     */
+    protected void decompressionBufferExhausted(ByteBuf buffer) {
+    }
+
 }

diff --git a/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
index 28f3919c604..015559ed3b7 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/JZlibTest.java
@@ -23,7 +23,7 @@ protected ZlibEncoder createEncoder(ZlibWrapper wrapper) {
     }
 
     @Override
-    protected ZlibDecoder createDecoder(ZlibWrapper wrapper) {
-        return new JZlibDecoder(wrapper);
+    protected ZlibDecoder createDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        return new JZlibDecoder(wrapper, maxAllocation);
     }
 }
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java
index 54a48a9caeb..5ff19f11540 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java
@@ -38,8 +38,8 @@ protected ZlibEncoder createEncoder(ZlibWrapper wrapper) {
     }
 
     @Override
-    protected ZlibDecoder createDecoder(ZlibWrapper wrapper) {
-        return new JdkZlibDecoder(wrapper);
+    protected ZlibDecoder createDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        return new JdkZlibDecoder(wrapper, maxAllocation);
     }
 
     @Test(expected = DecompressionException.class)
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest1.java b/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest1.java
index 9e16e1a3dcd..3c312742c85 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest1.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest1.java
@@ -23,7 +23,7 @@ protected ZlibEncoder createEncoder(ZlibWrapper wrapper) {
     }
 
     @Override
-    protected ZlibDecoder createDecoder(ZlibWrapper wrapper) {
-        return new JZlibDecoder(wrapper);
+    protected ZlibDecoder createDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        return new JZlibDecoder(wrapper, maxAllocation);
     }
 }
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest2.java b/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest2.java
index 8717019ed95..00c6e18c424 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest2.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/ZlibCrossTest2.java
@@ -25,8 +25,8 @@ protected ZlibEncoder createEncoder(ZlibWrapper wrapper) {
     }
 
     @Override
-    protected ZlibDecoder createDecoder(ZlibWrapper wrapper) {
-        return new JdkZlibDecoder(wrapper);
+    protected ZlibDecoder createDecoder(ZlibWrapper wrapper, int maxAllocation) {
+        return new JdkZlibDecoder(wrapper, maxAllocation);
     }
 
     @Test(expected = DecompressionException.class)
diff --git a/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java b/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java
index 24f76b2a810..5e9d1288fe8 100644
--- a/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/compression/ZlibTest.java
@@ -15,7 +15,9 @@
  */
 package io.netty.handler.codec.compression;
 
+import io.netty.buffer.AbstractByteBufAllocator;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufInputStream;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
@@ -88,8 +90,12 @@ public abstract class ZlibTest {
         rand.nextBytes(BYTES_LARGE);
     }
 
+    protected ZlibDecoder createDecoder(ZlibWrapper wrapper) {
+        return createDecoder(wrapper, 0);
+    }
+
     protected abstract ZlibEncoder createEncoder(ZlibWrapper wrapper);
-    protected abstract ZlibDecoder createDecoder(ZlibWrapper wrapper);
+    protected abstract ZlibDecoder createDecoder(ZlibWrapper wrapper, int maxAllocation);
 
     @Test
     public void testGZIP2() throws Exception {
@@ -345,6 +351,25 @@ public void testZLIB_OR_NONE3() throws Exception {
         testCompressLarge(ZlibWrapper.GZIP, ZlibWrapper.ZLIB_OR_NONE);
     }
 
+    @Test
+    public void testMaxAllocation() throws Exception {
+        int maxAllocation = 1024;
+        ZlibDecoder decoder = createDecoder(ZlibWrapper.ZLIB, maxAllocation);
+        EmbeddedChannel chDecoder = new EmbeddedChannel(decoder);
+        TestByteBufAllocator alloc = new TestByteBufAllocator(chDecoder.alloc());
+        chDecoder.config().setAllocator(alloc);
+
+        try {
+            chDecoder.writeInbound(Unpooled.wrappedBuffer(deflate(BYTES_LARGE)));
+            fail("decompressed size > maxAllocation, so should have thrown exception");
+        } catch (DecompressionException e) {
+            assertTrue(e.getMessage().startsWith("Decompression buffer has reached maximum size"));
+            assertEquals(maxAllocation, alloc.getMaxAllocation());
+            assertTrue(decoder.isClosed());
+            assertFalse(chDecoder.finish());
+        }
+    }
+
     private static byte[] gzip(byte[] bytes) throws IOException {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         GZIPOutputStream stream = new GZIPOutputStream(out);
@@ -360,4 +385,34 @@ private static byte[] deflate(byte[] bytes) throws IOException {
         stream.close();
         return out.toByteArray();
     }
+
+    private static final class TestByteBufAllocator extends AbstractByteBufAllocator {
+        private ByteBufAllocator wrapped;
+        private int maxAllocation;
+
+        TestByteBufAllocator(ByteBufAllocator wrapped) {
+            this.wrapped = wrapped;
+        }
+
+        public int getMaxAllocation() {
+            return maxAllocation;
+        }
+
+        @Override
+        public boolean isDirectBufferPooled() {
+            return wrapped.isDirectBufferPooled();
+        }
+
+        @Override
+        protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
+            maxAllocation = Math.max(maxAllocation, maxCapacity);
+            return wrapped.heapBuffer(initialCapacity, maxCapacity);
+        }
+
+        @Override
+        protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
+            maxAllocation = Math.max(maxAllocation, maxCapacity);
+            return wrapped.directBuffer(initialCapacity, maxCapacity);
+        }
+    }
 }
