Repository: netty/netty
Issue: ArrayIndexOutOfBoundsException With PooledByteBufAllocator Recycler implementaion

### Issue Description

While trying to write an HttpMessage using HTTP1 with HttpClientCodec i.e HttpObjectEncoder, Netty is throwing an ArrayIndexOutOfBounds Exception while trying to allocate buffer for headers. Exception seems to be happening in the  PooledByteBufAllocator Recycler Implementation.

```
Caused by: java.lang.ArrayIndexOutOfBoundsException: Index -841651488 out of bounds for length 256 
at io.netty.util.Recycler$Stack.pushNow(Recycler.java:666)  
at io.netty.util.Recycler$Stack.push(Recycler.java:642)  
at io.netty.util.Recycler$DefaultHandle.recycle(Recycler.java:236)  
at io.netty.buffer.PoolThreadCache$MemoryRegionCache$Entry.recycle(PoolThreadCache.java:482)     
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.allocate(PoolThreadCache.java:412)   
at io.netty.buffer.PoolThreadCache.allocate(PoolThreadCache.java:187)  
at io.netty.buffer.PoolThreadCache.allocateSmall(PoolThreadCache.java:171)  
at io.netty.buffer.PoolArena.allocate(PoolArena.java:189)    
at io.netty.buffer.PoolArena.allocate(PoolArena.java:147)  
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:356)  
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)    
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)  
at io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:115)  
at io.netty.handler.codec.http.HttpObjectEncoder.encode(HttpObjectEncoder.java:93)  
at io.netty.handler.codec.http.HttpClientCodec$Encoder.encode(HttpClientCodec.java:189)  
at io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:89)     ... 111 more
```

A new buffer is being allocated in Encoder here - https://github.com/netty/netty/blob/ee3b9a5f7b1829e1095fdbbccb5490949ac3e94e/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectEncoder.java#L93. 

Recycler code triggering the exception - https://github.com/netty/netty/blob/ee3b9a5f7b1829e1095fdbbccb5490949ac3e94e/common/src/main/java/io/netty/util/Recycler.java#L666

Similar issue seen with Recycler in the past - https://github.com/netty/netty/issues/9608

### Netty version
4.1.51.Final

### JVM version
Java 11

### OS version
x86_64 GNU/Linux

Issue comments:
Is this reproducible for you ?
Also I wonder if there was another exception before this as well .
We are not able to reproduce this behavior. There was no other exception thrown before this.
I was not able to spot any bug yet :/ 
We saw this issue re-occur with the same exact stack trace. There were no other exceptions logged before this. @normanmaurer are you looking to see some specific exception messaging from a specific file ? Also what is the log level for that line ?
Just share whatever exception was logged 
Since we're suspecting the Recycler is at fault here, I'm going to close this because the Recycler has been rewritten for Netty 4.1.71.
@chrisvest I am facing similar issue while PooledByteBuAllocator while writing Http/2 request

Attaching the stack trace

java.lang.ArrayIndexOutOfBoundsException: Index 39 out of bounds for length 39
	at io.netty.buffer.PoolArena.findSubpagePoolHead(PoolArena.java:267) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolChunk.free(PoolChunk.java:453) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolChunkList.free(PoolChunkList.java:120) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolArena.freeChunk(PoolArena.java:258) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache$MemoryRegionCache.freeEntry(PoolThreadCache.java:430) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:396) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache$MemoryRegionCache.trim(PoolThreadCache.java:414) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:277) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:269) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:258) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache.allocate(PoolThreadCache.java:170) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolThreadCache.allocateSmall(PoolThreadCache.java:151) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolArena.tcacheAllocateSmall(PoolArena.java:148) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolArena.allocate(PoolArena.java:134) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PoolArena.allocate(PoolArena.java:126) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:116) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DefaultHttp2FrameWriter.writeHeadersInternal(DefaultHttp2FrameWriter.java:515) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DefaultHttp2FrameWriter.writeHeaders(DefaultHttp2FrameWriter.java:260) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder.sendHeaders(DefaultHttp2ConnectionEncoder.java:184) ~[netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder.writeHeaders0(DefaultHttp2ConnectionEncoder.java:233) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder.writeHeaders(DefaultHttp2ConnectionEncoder.java:151) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.handler.codec.http2.DecoratingHttp2FrameWriter.writeHeaders(DecoratingHttp2FrameWriter.java:45) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at com.rt.jio.scp.http2.client.active.standby.ScpHttp2EgressActiveStandbyClientRequestHandler.write(ScpHttp2EgressActiveStandbyClientRequestHandler.java:118) [rtSDP_Run.jar:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:717) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:709) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.channel.AbstractChannelHandlerContext$WriteTask.run(AbstractChannelHandlerContext.java:1069) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:384) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-all-4.1.68.Final.jar:4.1.68.Final]
	at java.lang.Thread.run(Thread.java:832) [?:?]
@EzajAnsari Did this fix your issue?

@chrisvest When will 4.1.71 be released? I have been facing similar issue as @EzajAnsari for a couple of months and its blocking a production rollout. I am hoping [this](https://github.com/netty/netty/pull/11858) fixes the below issue too. 

```
java.lang.ArrayIndexOutOfBoundsException: 39, exiting...
io.netty.handler.codec.DecoderException: java.lang.ArrayIndexOutOfBoundsException: 39
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:478)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 39
at io.netty.buffer.PoolArena.findSubpagePoolHead(PoolArena.java:267)
at io.netty.buffer.PoolChunk.free(PoolChunk.java:453)
at io.netty.buffer.PoolChunkList.free(PoolChunkList.java:120)
at io.netty.buffer.PoolArena.freeChunk(PoolArena.java:258)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.freeEntry(PoolThreadCache.java:430)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:396)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.trim(PoolThreadCache.java:414)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:277)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:269)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:258)
at io.netty.buffer.PoolThreadCache.allocate(PoolThreadCache.java:170)
at io.netty.buffer.PoolThreadCache.allocateSmall(PoolThreadCache.java:151)
at io.netty.buffer.PoolArena.tcacheAllocateSmall(PoolArena.java:148)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:134)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:126)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)
at io.netty.handler.ssl.SslHandler.allocate(SslHandler.java:2194)
at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1402)
at io.netty.handler.ssl.SslHandler.decodeNonJdkCompatible(SslHandler.java:1265)
at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1302)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
... 16 more
```
@atanu1991 today 
> @EzajAnsari Did this fix your issue?
> 
> @chrisvest When will 4.1.71 be released? I have been facing similar issue as @EzajAnsari for a couple of months and its blocking a production rollout. I am hoping [this](https://github.com/netty/netty/pull/11858) fixes the below issue too.
> 
> ```
> java.lang.ArrayIndexOutOfBoundsException: 39, exiting...
> io.netty.handler.codec.DecoderException: java.lang.ArrayIndexOutOfBoundsException: 39
> at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:478)
> at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
> at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
> at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
> at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
> at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
> at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
> at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
> at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
> at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
> at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
> at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
> at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
> at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
> at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.lang.ArrayIndexOutOfBoundsException: 39
> at io.netty.buffer.PoolArena.findSubpagePoolHead(PoolArena.java:267)
> at io.netty.buffer.PoolChunk.free(PoolChunk.java:453)
> at io.netty.buffer.PoolChunkList.free(PoolChunkList.java:120)
> at io.netty.buffer.PoolArena.freeChunk(PoolArena.java:258)
> at io.netty.buffer.PoolThreadCache$MemoryRegionCache.freeEntry(PoolThreadCache.java:430)
> at io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:396)
> at io.netty.buffer.PoolThreadCache$MemoryRegionCache.trim(PoolThreadCache.java:414)
> at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:277)
> at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:269)
> at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:258)
> at io.netty.buffer.PoolThreadCache.allocate(PoolThreadCache.java:170)
> at io.netty.buffer.PoolThreadCache.allocateSmall(PoolThreadCache.java:151)
> at io.netty.buffer.PoolArena.tcacheAllocateSmall(PoolArena.java:148)
> at io.netty.buffer.PoolArena.allocate(PoolArena.java:134)
> at io.netty.buffer.PoolArena.allocate(PoolArena.java:126)
> at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395)
> at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
> at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)
> at io.netty.handler.ssl.SslHandler.allocate(SslHandler.java:2194)
> at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1402)
> at io.netty.handler.ssl.SslHandler.decodeNonJdkCompatible(SslHandler.java:1265)
> at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1302)
> at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:508)
> at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:447)
> ... 16 more
> ```

@chrisvest The error seems to still exist with 4.1.71 version of Netty. Any clue why this could be happening? Why is the number 39 (ArrayIndexOutOfBoundsException) always fixed?
@atanu1991 can you give us a stacktrace with 4.1.71.Final in use ?
@atanu1991 Also, if you could enable debug logging and collect the `-Dio.netty.allocator.*` lines, that would be helpful.
@chrisvest  @normanmaurer 

This is the log from 4.1.71.Final

```
java.lang.ArrayIndexOutOfBoundsException: 39
at io.netty.buffer.PoolArena.findSubpagePoolHead(PoolArena.java:267)
at io.netty.buffer.PoolChunk.free(PoolChunk.java:458)
at io.netty.buffer.PoolChunkList.free(PoolChunkList.java:120)
at io.netty.buffer.PoolArena.freeChunk(PoolArena.java:258)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.freeEntry(PoolThreadCache.java:430)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.free(PoolThreadCache.java:396)
at io.netty.buffer.PoolThreadCache$MemoryRegionCache.trim(PoolThreadCache.java:414)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:277)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:269)
at io.netty.buffer.PoolThreadCache.trim(PoolThreadCache.java:258)
at io.netty.buffer.PoolThreadCache.allocate(PoolThreadCache.java:170)
at io.netty.buffer.PoolThreadCache.allocateSmall(PoolThreadCache.java:151)
at io.netty.buffer.PoolArena.tcacheAllocateSmall(PoolArena.java:148)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:134)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:126)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:1031)
at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:826)
at io.netty.handler.ssl.SslHandler.wrapAndFlush(SslHandler.java:797)
at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:778)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:750)
at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:742)
at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:728)
at io.netty.handler.flush.FlushConsolidationHandler.flushNow(FlushConsolidationHandler.java:204)
at io.netty.handler.flush.FlushConsolidationHandler.flushIfNeeded(FlushConsolidationHandler.java:197)
at io.netty.handler.flush.FlushConsolidationHandler.resetReadAndFlushIfNeeded(FlushConsolidationHandler.java:192)
at io.netty.handler.flush.FlushConsolidationHandler.channelReadComplete(FlushConsolidationHandler.java:145)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
at io.netty.handler.ssl.SslHandler.channelReadComplete0(SslHandler.java:1302)
at io.netty.handler.ssl.SslHandler.channelReadComplete(SslHandler.java:1291)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelReadComplete(DefaultChannelPipeline.java:1415)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
at io.netty.channel.DefaultChannelPipeline.fireChannelReadComplete(DefaultChannelPipeline.java:925)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:171)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at java.lang.Thread.run(Thread.java:748)
```

For context we are using Netty from Apache Giraph: https://github.com/apache/giraph
Another point to note is that this is flaky and the same job sometimes pass and sometimes throws the above exception and fail.

I did start our service in debug mode but I dont see io.netty.allocator log lines. Is there some special command I need to pass?
@chrisvest  @normanmaurer 

Any thoughts on this?
@atanu1991 dint have time yet. Are you able to share a reproducer ?
https://github.com/netty/netty/pull/11939

Pull request review comments:
why not using a mpsc unbounded queue from JCTools here?

`claim()` is supposed to be called by the same thread (and its corresponding drain too), while filling it could happen from several threads; it looks an easy way to save using synchronization


draining inbox would worth if there isn't anything on pooledHandles so we can defer draining until is necessary or if the external mailbox got "too big"?
Good call. This simplifies the code even more. Performance is roughly unchanged in my measurements.
No longer relevant.
I would expose chunk size of the queue: it impacts the footprint in memory while the pool is in steady state ie contains < next power of 2 chunk size  elements

https://github.com/JCTools/JCTools/blob/master/jctools-core/src/main/java/org/jctools/queues/MpscGrowableArrayQueue.java#L28 is a good fit as well: it would left the chunk capacity to change accordly to the load (max reached capacity) until it would be stable; but meaning that its footprint will be the max capacity ever reached.
Nice one: if the release vs acquire rate isn't that high it shouldn't indeed, but if there are many interleaving release/acquire yep because of the contention while draining from inbox vs releasing to it
nit: why not just create this directly ? I think there is really no need for the extra static block here. 
Because that would require two casts in a row in order to get the generics to line up. And that would look ugly. We've done static blocks for this situation in other places, so it's not a new pattern. 

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/Recycler.java b/common/src/main/java/io/netty/util/Recycler.java
index eec543f2d22..373a79251b4 100644
--- a/common/src/main/java/io/netty/util/Recycler.java
+++ b/common/src/main/java/io/netty/util/Recycler.java
@@ -13,23 +13,18 @@
  * License for the specific language governing permissions and limitations
  * under the License.
  */
-
 package io.netty.util;
 
 import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.internal.ObjectPool;
+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.lang.ref.WeakReference;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.WeakHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.Queue;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
-import static io.netty.util.internal.MathUtil.safeFindNextPositivePowerOfTwo;
 import static java.lang.Math.max;
 import static java.lang.Math.min;
 
@@ -39,26 +34,22 @@
  * @param <T> the type of the pooled object
  */
 public abstract class Recycler<T> {
-
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(Recycler.class);
-
-    @SuppressWarnings("rawtypes")
-    private static final Handle NOOP_HANDLE = new Handle() {
+    private static final Handle<?> NOOP_HANDLE = new Handle<Object>() {
         @Override
         public void recycle(Object object) {
             // NOOP
         }
+
+        @Override
+        public String toString() {
+            return "NOOP_HANDLE";
+        }
     };
-    private static final AtomicInteger ID_GENERATOR = new AtomicInteger(Integer.MIN_VALUE);
-    private static final int OWN_THREAD_ID = ID_GENERATOR.getAndIncrement();
     private static final int DEFAULT_INITIAL_MAX_CAPACITY_PER_THREAD = 4 * 1024; // Use 4k instances as default.
     private static final int DEFAULT_MAX_CAPACITY_PER_THREAD;
-    private static final int INITIAL_CAPACITY;
-    private static final int MAX_SHARED_CAPACITY_FACTOR;
-    private static final int MAX_DELAYED_QUEUES_PER_THREAD;
-    private static final int LINK_CAPACITY;
     private static final int RATIO;
-    private static final int DELAYED_QUEUE_RATIO;
+    private static final int DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD;
 
     static {
         // In the future, we might have different maxCapacity for different object types.
@@ -71,65 +62,33 @@ public void recycle(Object object) {
         }
 
         DEFAULT_MAX_CAPACITY_PER_THREAD = maxCapacityPerThread;
-
-        MAX_SHARED_CAPACITY_FACTOR = max(2,
-                SystemPropertyUtil.getInt("io.netty.recycler.maxSharedCapacityFactor",
-                        2));
-
-        MAX_DELAYED_QUEUES_PER_THREAD = max(0,
-                SystemPropertyUtil.getInt("io.netty.recycler.maxDelayedQueuesPerThread",
-                        // We use the same value as default EventLoop number
-                        NettyRuntime.availableProcessors() * 2));
-
-        LINK_CAPACITY = safeFindNextPositivePowerOfTwo(
-                max(SystemPropertyUtil.getInt("io.netty.recycler.linkCapacity", 16), 16));
+        DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD = SystemPropertyUtil.getInt("io.netty.recycler.chunkSize", 32);
 
         // By default we allow one push to a Recycler for each 8th try on handles that were never recycled before.
         // This should help to slowly increase the capacity of the recycler while not be too sensitive to allocation
         // bursts.
         RATIO = max(0, SystemPropertyUtil.getInt("io.netty.recycler.ratio", 8));
-        DELAYED_QUEUE_RATIO = max(0, SystemPropertyUtil.getInt("io.netty.recycler.delayedQueue.ratio", RATIO));
-
-        INITIAL_CAPACITY = min(DEFAULT_MAX_CAPACITY_PER_THREAD, 256);
 
         if (logger.isDebugEnabled()) {
             if (DEFAULT_MAX_CAPACITY_PER_THREAD == 0) {
                 logger.debug("-Dio.netty.recycler.maxCapacityPerThread: disabled");
-                logger.debug("-Dio.netty.recycler.maxSharedCapacityFactor: disabled");
-                logger.debug("-Dio.netty.recycler.linkCapacity: disabled");
                 logger.debug("-Dio.netty.recycler.ratio: disabled");
-                logger.debug("-Dio.netty.recycler.delayedQueue.ratio: disabled");
+                logger.debug("-Dio.netty.recycler.chunkSize: disabled");
             } else {
                 logger.debug("-Dio.netty.recycler.maxCapacityPerThread: {}", DEFAULT_MAX_CAPACITY_PER_THREAD);
-                logger.debug("-Dio.netty.recycler.maxSharedCapacityFactor: {}", MAX_SHARED_CAPACITY_FACTOR);
-                logger.debug("-Dio.netty.recycler.linkCapacity: {}", LINK_CAPACITY);
                 logger.debug("-Dio.netty.recycler.ratio: {}", RATIO);
-                logger.debug("-Dio.netty.recycler.delayedQueue.ratio: {}", DELAYED_QUEUE_RATIO);
+                logger.debug("-Dio.netty.recycler.chunkSize: {}", DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD);
             }
         }
     }
 
     private final int maxCapacityPerThread;
-    private final int maxSharedCapacityFactor;
     private final int interval;
-    private final int maxDelayedQueuesPerThread;
-    private final int delayedQueueInterval;
-
-    private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {
+    private final int chunkSize;
+    private final FastThreadLocal<LocalPool<T>> threadLocal = new FastThreadLocal<LocalPool<T>>() {
         @Override
-        protected Stack<T> initialValue() {
-            return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,
-                    interval, maxDelayedQueuesPerThread, delayedQueueInterval);
-        }
-
-        @Override
-        protected void onRemoval(Stack<T> value) {
-            // Let us remove the WeakOrderQueue from the WeakHashMap directly if its safe to remove some overhead
-            if (value.threadRef.get() == Thread.currentThread()) {
-               if (DELAYED_RECYCLED.isSet()) {
-                   DELAYED_RECYCLED.get().remove(value);
-               }
-            }
+        protected LocalPool<T> initialValue() {
+            return new LocalPool<T>(maxCapacityPerThread, interval, chunkSize);
         }
     };
 
@@ -138,31 +97,49 @@ protected Recycler() {
     }
 
     protected Recycler(int maxCapacityPerThread) {
-        this(maxCapacityPerThread, MAX_SHARED_CAPACITY_FACTOR);
+        this(maxCapacityPerThread, RATIO, DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD);
     }
 
+    /**
+     * @deprecated Use one of the following instead:
+     * {@link #Recycler()}, {@link #Recycler(int)}, {@link #Recycler(int, int, int)}.
+     */
+    @Deprecated
+    @SuppressWarnings("unused") // Parameters we can't remove due to compatibility.
     protected Recycler(int maxCapacityPerThread, int maxSharedCapacityFactor) {
-        this(maxCapacityPerThread, maxSharedCapacityFactor, RATIO, MAX_DELAYED_QUEUES_PER_THREAD);
+        this(maxCapacityPerThread, RATIO, DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD);
     }
 
+    /**
+     * @deprecated Use one of the following instead:
+     * {@link #Recycler()}, {@link #Recycler(int)}, {@link #Recycler(int, int, int)}.
+     */
+    @Deprecated
+    @SuppressWarnings("unused") // Parameters we can't remove due to compatibility.
     protected Recycler(int maxCapacityPerThread, int maxSharedCapacityFactor,
                        int ratio, int maxDelayedQueuesPerThread) {
-        this(maxCapacityPerThread, maxSharedCapacityFactor, ratio, maxDelayedQueuesPerThread,
-                DELAYED_QUEUE_RATIO);
+        this(maxCapacityPerThread, ratio, DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD);
     }
 
+    /**
+     * @deprecated Use one of the following instead:
+     * {@link #Recycler()}, {@link #Recycler(int)}, {@link #Recycler(int, int, int)}.
+     */
+    @Deprecated
+    @SuppressWarnings("unused") // Parameters we can't remove due to compatibility.
     protected Recycler(int maxCapacityPerThread, int maxSharedCapacityFactor,
                        int ratio, int maxDelayedQueuesPerThread, int delayedQueueRatio) {
+        this(maxCapacityPerThread, ratio, DEFAULT_QUEUE_CHUNK_SIZE_PER_THREAD);
+    }
+
+    protected Recycler(int maxCapacityPerThread, int ratio, int chunkSize) {
         interval = max(0, ratio);
-        delayedQueueInterval = max(0, delayedQueueRatio);
         if (maxCapacityPerThread <= 0) {
             this.maxCapacityPerThread = 0;
-            this.maxSharedCapacityFactor = 1;
-            this.maxDelayedQueuesPerThread = 0;
+            this.chunkSize = 0;
         } else {
-            this.maxCapacityPerThread = maxCapacityPerThread;
-            this.maxSharedCapacityFactor = max(1, maxSharedCapacityFactor);
-            this.maxDelayedQueuesPerThread = max(0, maxDelayedQueuesPerThread);
+            this.maxCapacityPerThread = max(4, maxCapacityPerThread);
+            this.chunkSize = max(2, min(chunkSize, this.maxCapacityPerThread >> 1));
         }
     }
 
@@ -171,13 +148,22 @@ public final T get() {
         if (maxCapacityPerThread == 0) {
             return newObject((Handle<T>) NOOP_HANDLE);
         }
-        Stack<T> stack = threadLocal.get();
-        DefaultHandle<T> handle = stack.pop();
+        LocalPool<T> localPool = threadLocal.get();
+        DefaultHandle<T> handle = localPool.claim();
+        T obj;
         if (handle == null) {
-            handle = stack.newHandle();
-            handle.value = newObject(handle);
+            handle = localPool.newHandle();
+            if (handle != null) {
+                obj = newObject(handle);
+                handle.set(obj);
+            } else {
+                obj = newObject((Handle<T>) NOOP_HANDLE);
+            }
+        } else {
+            obj = handle.get();
         }
-        return (T) handle.value;
+
+        return obj;
     }
 
     /**
@@ -189,46 +175,36 @@ public final boolean recycle(T o, Handle<T> handle) {
             return false;
         }
 
-        DefaultHandle<T> h = (DefaultHandle<T>) handle;
-        if (h.stack.parent != this) {
-            return false;
-        }
-
-        h.recycle(o);
+        handle.recycle(o);
         return true;
     }
 
-    final int threadLocalCapacity() {
-        return threadLocal.get().elements.length;
-    }
-
     final int threadLocalSize() {
-        return threadLocal.get().size;
+        return threadLocal.get().pooledHandles.size();
     }
 
     protected abstract T newObject(Handle<T> handle);
 
+    @SuppressWarnings("ClassNameSameAsAncestorName") // Can't change this due to compatibility.
     public interface Handle<T> extends ObjectPool.Handle<T>  { }
 
-    @SuppressWarnings("unchecked")
     private static final class DefaultHandle<T> implements Handle<T> {
-        private static final AtomicIntegerFieldUpdater<DefaultHandle<?>> LAST_RECYCLED_ID_UPDATER;
+        private static final int STATE_CLAIMED = 0;
+        private static final int STATE_AVAILABLE = 1;
+        private static final AtomicIntegerFieldUpdater<DefaultHandle<?>> STATE_UPDATER;
         static {
-            AtomicIntegerFieldUpdater<?> updater = AtomicIntegerFieldUpdater.newUpdater(
-                    DefaultHandle.class, "lastRecycledId");
-            LAST_RECYCLED_ID_UPDATER = (AtomicIntegerFieldUpdater<DefaultHandle<?>>) updater;
+            AtomicIntegerFieldUpdater<?> updater = AtomicIntegerFieldUpdater.newUpdater(DefaultHandle.class, "state");
+            //noinspection unchecked
+            STATE_UPDATER = (AtomicIntegerFieldUpdater<DefaultHandle<?>>) updater;
         }
 
-        volatile int lastRecycledId;
-        int recycleId;
-
-        boolean hasBeenRecycled;
-
-        Stack<?> stack;
-        Object value;
+        @SuppressWarnings({"FieldMayBeFinal", "unused"}) // Updated by STATE_UPDATER.
+        private volatile int state; // State is initialised to STATE_CLAIMED (aka. 0) so they can be released.
+        private final LocalPool<T> localPool;
+        private T value;
 
-        DefaultHandle(Stack<?> stack) {
-            this.stack = stack;
+        DefaultHandle(LocalPool<T> localPool) {
+            this.localPool = localPool;
         }
 
         @Override
@@ -236,511 +212,63 @@ public void recycle(Object object) {
             if (object != value) {
                 throw new IllegalArgumentException("object does not belong to handle");
             }
-
-            Stack<?> stack = this.stack;
-            if (lastRecycledId != recycleId || stack == null) {
-                throw new IllegalStateException("recycled already");
-            }
-
-            stack.push(this);
-        }
-
-        public boolean compareAndSetLastRecycledId(int expectLastRecycledId, int updateLastRecycledId) {
-            // Use "weak…" because we do not need synchronize-with ordering, only atomicity.
-            // Also, spurious failures are fine, since no code should rely on recycling for correctness.
-            return LAST_RECYCLED_ID_UPDATER.weakCompareAndSet(this, expectLastRecycledId, updateLastRecycledId);
+            localPool.release(this);
         }
-    }
 
-    private static final FastThreadLocal<Map<Stack<?>, WeakOrderQueue>> DELAYED_RECYCLED =
-            new FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>() {
-        @Override
-        protected Map<Stack<?>, WeakOrderQueue> initialValue() {
-            return new WeakHashMap<Stack<?>, WeakOrderQueue>();
+        T get() {
+            return value;
         }
-    };
-
-    // a queue that makes only moderate guarantees about visibility: items are seen in the correct order,
-    // but we aren't absolutely guaranteed to ever see anything at all, thereby keeping the queue cheap to maintain
-    private static final class WeakOrderQueue extends WeakReference<Thread> {
-
-        static final WeakOrderQueue DUMMY = new WeakOrderQueue();
-
-        // Let Link extend AtomicInteger for intrinsics. The Link itself will be used as writerIndex.
-        @SuppressWarnings("serial")
-        static final class Link extends AtomicInteger {
-            final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];
 
-            int readIndex;
-            Link next;
+        void set(T value) {
+            this.value = value;
         }
 
-        // Its important this does not hold any reference to either Stack or WeakOrderQueue.
-        private static final class Head {
-            private final AtomicInteger availableSharedCapacity;
-
-            Link link;
-
-            Head(AtomicInteger availableSharedCapacity) {
-                this.availableSharedCapacity = availableSharedCapacity;
-            }
-
-            /**
-             * Reclaim all used space and also unlink the nodes to prevent GC nepotism.
-             */
-            void reclaimAllSpaceAndUnlink() {
-                Link head = link;
-                link = null;
-                int reclaimSpace = 0;
-                while (head != null) {
-                    reclaimSpace += LINK_CAPACITY;
-                    Link next = head.next;
-                    // Unlink to help GC and guard against GC nepotism.
-                    head.next = null;
-                    head = next;
-                }
-                if (reclaimSpace > 0) {
-                    reclaimSpace(reclaimSpace);
-                }
-            }
-
-            private void reclaimSpace(int space) {
-                availableSharedCapacity.addAndGet(space);
-            }
-
-            void relink(Link link) {
-                reclaimSpace(LINK_CAPACITY);
-                this.link = link;
-            }
-
-            /**
-             * Creates a new {@link} and returns it if we can reserve enough space for it, otherwise it
-             * returns {@code null}.
-             */
-            Link newLink() {
-                return reserveSpaceForLink(availableSharedCapacity) ? new Link() : null;
-            }
-
-            static boolean reserveSpaceForLink(AtomicInteger availableSharedCapacity) {
-                for (;;) {
-                    int available = availableSharedCapacity.get();
-                    if (available < LINK_CAPACITY) {
-                        return false;
-                    }
-                    if (availableSharedCapacity.compareAndSet(available, available - LINK_CAPACITY)) {
-                        return true;
-                    }
-                }
-            }
-        }
-
-        // chain of data items
-        private final Head head;
-        private Link tail;
-        // pointer to another queue of delayed items for the same stack
-        private WeakOrderQueue next;
-        private final int id = ID_GENERATOR.getAndIncrement();
-        private final int interval;
-        private int handleRecycleCount;
-
-        private WeakOrderQueue() {
-            super(null);
-            head = new Head(null);
-            interval = 0;
-        }
-
-        private WeakOrderQueue(Stack<?> stack, Thread thread) {
-            super(thread);
-            tail = new Link();
-
-            // Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in
-            // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the
-            // Stack itself GCed.
-            head = new Head(stack.availableSharedCapacity);
-            head.link = tail;
-            interval = stack.delayedQueueInterval;
-            handleRecycleCount = interval; // Start at interval so the first one will be recycled.
-        }
-
-        static WeakOrderQueue newQueue(Stack<?> stack, Thread thread) {
-            // We allocated a Link so reserve the space
-            if (!Head.reserveSpaceForLink(stack.availableSharedCapacity)) {
-                return null;
-            }
-            final WeakOrderQueue queue = new WeakOrderQueue(stack, thread);
-            // Done outside of the constructor to ensure WeakOrderQueue.this does not escape the constructor and so
-            // may be accessed while its still constructed.
-            stack.setHead(queue);
-
-            return queue;
-        }
-
-        WeakOrderQueue getNext() {
-            return next;
-        }
-
-        void setNext(WeakOrderQueue next) {
-            assert next != this;
-            this.next = next;
-        }
-
-        void reclaimAllSpaceAndUnlink() {
-            head.reclaimAllSpaceAndUnlink();
-            next = null;
-        }
-
-        void add(DefaultHandle<?> handle) {
-            if (!handle.compareAndSetLastRecycledId(0, id)) {
-                // Separate threads could be racing to add the handle to each their own WeakOrderQueue.
-                // We only add the handle to the queue if we win the race and observe that lastRecycledId is zero.
-                return;
-            }
-
-            // While we also enforce the recycling ratio when we transfer objects from the WeakOrderQueue to the Stack
-            // we better should enforce it as well early. Missing to do so may let the WeakOrderQueue grow very fast
-            // without control
-            if (!handle.hasBeenRecycled) {
-                if (handleRecycleCount < interval) {
-                    handleRecycleCount++;
-                    // Drop the item to prevent from recycling too aggressively.
-                    return;
-                }
-                handleRecycleCount = 0;
-            }
-
-            Link tail = this.tail;
-            int writeIndex;
-            if ((writeIndex = tail.get()) == LINK_CAPACITY) {
-                Link link = head.newLink();
-                if (link == null) {
-                    // Drop it.
-                    return;
-                }
-                // We allocate a Link so reserve the space
-                this.tail = tail = tail.next = link;
-
-                writeIndex = tail.get();
-            }
-            tail.elements[writeIndex] = handle;
-            handle.stack = null;
-            // we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;
-            // this also means we guarantee visibility of an element in the queue if we see the index updated
-            tail.lazySet(writeIndex + 1);
-        }
-
-        boolean hasFinalData() {
-            return tail.readIndex != tail.get();
-        }
-
-        // transfer as many items as we can from this queue to the stack, returning true if any were transferred
-        @SuppressWarnings("rawtypes")
-        boolean transfer(Stack<?> dst) {
-            Link head = this.head.link;
-            if (head == null) {
+        boolean availableToClaim() {
+            if (state != STATE_AVAILABLE) {
                 return false;
             }
-
-            if (head.readIndex == LINK_CAPACITY) {
-                if (head.next == null) {
-                    return false;
-                }
-                head = head.next;
-                this.head.relink(head);
-            }
-
-            final int srcStart = head.readIndex;
-            int srcEnd = head.get();
-            final int srcSize = srcEnd - srcStart;
-            if (srcSize == 0) {
-                return false;
-            }
-
-            final int dstSize = dst.size;
-            final int expectedCapacity = dstSize + srcSize;
-
-            if (expectedCapacity > dst.elements.length) {
-                final int actualCapacity = dst.increaseCapacity(expectedCapacity);
-                srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);
-            }
-
-            if (srcStart != srcEnd) {
-                final DefaultHandle[] srcElems = head.elements;
-                final DefaultHandle[] dstElems = dst.elements;
-                int newDstSize = dstSize;
-                for (int i = srcStart; i < srcEnd; i++) {
-                    DefaultHandle<?> element = srcElems[i];
-                    if (element.recycleId == 0) {
-                        element.recycleId = element.lastRecycledId;
-                    } else if (element.recycleId != element.lastRecycledId) {
-                        throw new IllegalStateException("recycled already");
-                    }
-                    srcElems[i] = null;
-
-                    if (dst.dropHandle(element)) {
-                        // Drop the object.
-                        continue;
-                    }
-                    element.stack = dst;
-                    dstElems[newDstSize ++] = element;
-                }
-
-                if (srcEnd == LINK_CAPACITY && head.next != null) {
-                    // Add capacity back as the Link is GCed.
-                    this.head.relink(head.next);
-                }
-
-                head.readIndex = srcEnd;
-                if (dst.size == newDstSize) {
-                    return false;
-                }
-                dst.size = newDstSize;
-                return true;
-            } else {
-                // The destination stack is full already.
-                return false;
-            }
-        }
-    }
-
-    private static final class Stack<T> {
-
-        // we keep a queue of per-thread queues, which is appended to once only, each time a new thread other
-        // than the stack owner recycles: when we run out of items in our stack we iterate this collection
-        // to scavenge those that can be reused. this permits us to incur minimal thread synchronisation whilst
-        // still recycling all items.
-        final Recycler<T> parent;
-
-        // We store the Thread in a WeakReference as otherwise we may be the only ones that still hold a strong
-        // Reference to the Thread itself after it died because DefaultHandle will hold a reference to the Stack.
-        //
-        // The biggest issue is if we do not use a WeakReference the Thread may not be able to be collected at all if
-        // the user will store a reference to the DefaultHandle somewhere and never clear this reference (or not clear
-        // it in a timely manner).
-        final WeakReference<Thread> threadRef;
-        final AtomicInteger availableSharedCapacity;
-        private final int maxDelayedQueues;
-
-        private final int maxCapacity;
-        private final int interval;
-        private final int delayedQueueInterval;
-        DefaultHandle<?>[] elements;
-        int size;
-        private int handleRecycleCount;
-        private WeakOrderQueue cursor, prev;
-        private volatile WeakOrderQueue head;
-
-        Stack(Recycler<T> parent, Thread thread, int maxCapacity, int maxSharedCapacityFactor,
-              int interval, int maxDelayedQueues, int delayedQueueInterval) {
-            this.parent = parent;
-            threadRef = new WeakReference<Thread>(thread);
-            this.maxCapacity = maxCapacity;
-            availableSharedCapacity = new AtomicInteger(max(maxCapacity / maxSharedCapacityFactor, LINK_CAPACITY));
-            elements = new DefaultHandle[min(INITIAL_CAPACITY, maxCapacity)];
-            this.interval = interval;
-            this.delayedQueueInterval = delayedQueueInterval;
-            handleRecycleCount = interval; // Start at interval so the first one will be recycled.
-            this.maxDelayedQueues = maxDelayedQueues;
-        }
-
-        // Marked as synchronized to ensure this is serialized.
-        synchronized void setHead(WeakOrderQueue queue) {
-            queue.setNext(head);
-            head = queue;
+            return STATE_UPDATER.compareAndSet(this, STATE_AVAILABLE, STATE_CLAIMED);
         }
 
-        int increaseCapacity(int expectedCapacity) {
-            int newCapacity = elements.length;
-            int maxCapacity = this.maxCapacity;
-            do {
-                newCapacity <<= 1;
-            } while (newCapacity < expectedCapacity && newCapacity < maxCapacity);
-
-            newCapacity = min(newCapacity, maxCapacity);
-            if (newCapacity != elements.length) {
-                elements = Arrays.copyOf(elements, newCapacity);
+        void toAvailable() {
+            int prev = STATE_UPDATER.getAndSet(this, STATE_AVAILABLE);
+            if (prev == STATE_AVAILABLE) {
+                throw new IllegalStateException("Object has been recycled already.");
             }
-
-            return newCapacity;
-        }
-
-        @SuppressWarnings({ "unchecked", "rawtypes" })
-        DefaultHandle<T> pop() {
-            int size = this.size;
-            if (size == 0) {
-                if (!scavenge()) {
-                    return null;
-                }
-                size = this.size;
-                if (size <= 0) {
-                    // double check, avoid races
-                    return null;
-                }
-            }
-            size --;
-            DefaultHandle ret = elements[size];
-            elements[size] = null;
-            // As we already set the element[size] to null we also need to store the updated size before we do
-            // any validation. Otherwise we may see a null value when later try to pop again without a new element
-            // added before.
-            this.size = size;
-
-            if (ret.lastRecycledId != ret.recycleId) {
-                throw new IllegalStateException("recycled multiple times");
-            }
-            ret.recycleId = 0;
-            ret.lastRecycledId = 0;
-            return ret;
         }
+    }
 
-        private boolean scavenge() {
-            // continue an existing scavenge, if any
-            if (scavengeSome()) {
-                return true;
-            }
+    private static final class LocalPool<T> {
+        private final int ratioInterval;
+        private final Queue<DefaultHandle<T>> pooledHandles;
+        private int ratioCounter;
 
-            // reset our scavenge cursor
-            prev = null;
-            cursor = head;
-            return false;
+        LocalPool(int maxCapacity, int ratioInterval, int chunkSize) {
+            this.ratioInterval = ratioInterval;
+            pooledHandles = PlatformDependent.newMpscQueue(chunkSize, maxCapacity);
+            ratioCounter = ratioInterval; // Start at interval so the first one will be recycled.
         }
 
-        private boolean scavengeSome() {
-            WeakOrderQueue prev;
-            WeakOrderQueue cursor = this.cursor;
-            if (cursor == null) {
-                prev = null;
-                cursor = head;
-                if (cursor == null) {
-                    return false;
-                }
-            } else {
-                prev = this.prev;
-            }
-
-            boolean success = false;
+        DefaultHandle<T> claim() {
+            Queue<DefaultHandle<T>> pooledHandles = this.pooledHandles;
+            DefaultHandle<T> handle;
             do {
-                if (cursor.transfer(this)) {
-                    success = true;
-                    break;
-                }
-                WeakOrderQueue next = cursor.getNext();
-                if (cursor.get() == null) {
-                    // If the thread associated with the queue is gone, unlink it, after
-                    // performing a volatile read to confirm there is no data left to collect.
-                    // We never unlink the first queue, as we don't want to synchronize on updating the head.
-                    if (cursor.hasFinalData()) {
-                        for (;;) {
-                            if (cursor.transfer(this)) {
-                                success = true;
-                            } else {
-                                break;
-                            }
-                        }
-                    }
-
-                    if (prev != null) {
-                        // Ensure we reclaim all space before dropping the WeakOrderQueue to be GC'ed.
-                        cursor.reclaimAllSpaceAndUnlink();
-                        prev.setNext(next);
-                    }
-                } else {
-                    prev = cursor;
-                }
-
-                cursor = next;
-
-            } while (cursor != null && !success);
-
-            this.prev = prev;
-            this.cursor = cursor;
-            return success;
-        }
-
-        void push(DefaultHandle<?> item) {
-            Thread currentThread = Thread.currentThread();
-            if (threadRef.get() == currentThread) {
-                // The current Thread is the thread that belongs to the Stack, we can try to push the object now.
-                pushNow(item);
-            } else {
-                // The current Thread is not the one that belongs to the Stack
-                // (or the Thread that belonged to the Stack was collected already), we need to signal that the push
-                // happens later.
-                pushLater(item, currentThread);
-            }
-        }
-
-        private void pushNow(DefaultHandle<?> item) {
-            if (item.recycleId != 0 || !item.compareAndSetLastRecycledId(0, OWN_THREAD_ID)) {
-                throw new IllegalStateException("recycled already");
-            }
-            item.recycleId = OWN_THREAD_ID;
-
-            int size = this.size;
-            if (size >= maxCapacity || dropHandle(item)) {
-                // Hit the maximum capacity or should drop - drop the possibly youngest object.
-                return;
-            }
-            if (size == elements.length) {
-                elements = Arrays.copyOf(elements, min(size << 1, maxCapacity));
-            }
-
-            elements[size] = item;
-            this.size = size + 1;
+                handle = pooledHandles.poll();
+            } while (handle != null && !handle.availableToClaim());
+            return handle;
         }
 
-        private void pushLater(DefaultHandle<?> item, Thread thread) {
-            if (maxDelayedQueues == 0) {
-                // We don't support recycling across threads and should just drop the item on the floor.
-                return;
-            }
-
-            // we don't want to have a ref to the queue as the value in our weak map
-            // so we null it out; to ensure there are no races with restoring it later
-            // we impose a memory ordering here (no-op on x86)
-            Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
-            WeakOrderQueue queue = delayedRecycled.get(this);
-            if (queue == null) {
-                if (delayedRecycled.size() >= maxDelayedQueues) {
-                    // Add a dummy queue so we know we should drop the object
-                    delayedRecycled.put(this, WeakOrderQueue.DUMMY);
-                    return;
-                }
-                // Check if we already reached the maximum number of delayed queues and if we can allocate at all.
-                if ((queue = newWeakOrderQueue(thread)) == null) {
-                    // drop object
-                    return;
-                }
-                delayedRecycled.put(this, queue);
-            } else if (queue == WeakOrderQueue.DUMMY) {
-                // drop object
-                return;
-            }
-
-            queue.add(item);
-        }
-
-        /**
-         * Allocate a new {@link WeakOrderQueue} or return {@code null} if not possible.
-         */
-        private WeakOrderQueue newWeakOrderQueue(Thread thread) {
-            return WeakOrderQueue.newQueue(this, thread);
-        }
-
-        boolean dropHandle(DefaultHandle<?> handle) {
-            if (!handle.hasBeenRecycled) {
-                if (handleRecycleCount < interval) {
-                    handleRecycleCount++;
-                    // Drop the object.
-                    return true;
-                }
-                handleRecycleCount = 0;
-                handle.hasBeenRecycled = true;
-            }
-            return false;
+        void release(DefaultHandle<T> handle) {
+            handle.toAvailable();
+            pooledHandles.offer(handle);
         }
 
         DefaultHandle<T> newHandle() {
-            return new DefaultHandle<T>(this);
+            if (++ratioCounter >= ratioInterval) {
+                ratioCounter = 0;
+                return new DefaultHandle<T>(this);
+            }
+            return null;
         }
     }
 }
diff --git a/common/src/main/java/io/netty/util/internal/PlatformDependent.java b/common/src/main/java/io/netty/util/internal/PlatformDependent.java
index 099aba86765..a116da2451f 100644
--- a/common/src/main/java/io/netty/util/internal/PlatformDependent.java
+++ b/common/src/main/java/io/netty/util/internal/PlatformDependent.java
@@ -974,8 +974,12 @@ static <T> Queue<T> newMpscQueue(final int maxCapacity) {
             // This is forced by the MpscChunkedArrayQueue implementation as will try to round it
             // up to the next power of two and so will overflow otherwise.
             final int capacity = max(min(maxCapacity, MAX_ALLOWED_MPSC_CAPACITY), MIN_MAX_MPSC_CAPACITY);
-            return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscChunkedArrayQueue<T>(MPSC_CHUNK_SIZE, capacity)
-                                                : new MpscChunkedAtomicArrayQueue<T>(MPSC_CHUNK_SIZE, capacity);
+            return newChunkedMpscQueue(MPSC_CHUNK_SIZE, capacity);
+        }
+
+        static <T> Queue<T> newChunkedMpscQueue(final int chunkSize, final int capacity) {
+            return USE_MPSC_CHUNKED_ARRAY_QUEUE ? new MpscChunkedArrayQueue<T>(chunkSize, capacity)
+                    : new MpscChunkedAtomicArrayQueue<T>(chunkSize, capacity);
         }
 
         static <T> Queue<T> newMpscQueue() {
@@ -1001,6 +1005,15 @@ public static <T> Queue<T> newMpscQueue(final int maxCapacity) {
         return Mpsc.newMpscQueue(maxCapacity);
     }
 
+    /**
+     * Create a new {@link Queue} which is safe to use for multiple producers (different threads) and a single
+     * consumer (one thread!).
+     * The queue will grow and shrink its capacity in units of the given chunk size.
+     */
+    public static <T> Queue<T> newMpscQueue(final int chunkSize, final int maxCapacity) {
+        return Mpsc.newChunkedMpscQueue(chunkSize, maxCapacity);
+    }
+
     /**
      * Create a new {@link Queue} which is safe to use for single producer (one thread!) and a single
      * consumer (one thread!).
diff --git a/microbench/src/main/java/io/netty/microbench/util/RecyclerBenchmark.java b/microbench/src/main/java/io/netty/microbench/util/RecyclerBenchmark.java
new file mode 100644
index 00000000000..eb02774dbb6
--- /dev/null
+++ b/microbench/src/main/java/io/netty/microbench/util/RecyclerBenchmark.java
@@ -0,0 +1,125 @@
+/*
+ * 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.microbench.util;
+
+import io.netty.util.Recycler;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Control;
+import org.openjdk.jmh.runner.options.ChainedOptionsBuilder;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+@Warmup(iterations = AbstractMicrobenchmarkBase.DEFAULT_WARMUP_ITERATIONS, time = 1)
+@Measurement(iterations = AbstractMicrobenchmarkBase.DEFAULT_MEASURE_ITERATIONS, time = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+public class RecyclerBenchmark extends AbstractMicrobenchmark {
+    private Recycler<DummyObject> recycler = new Recycler<DummyObject>() {
+        @Override
+        protected DummyObject newObject(Recycler.Handle<DummyObject> handle) {
+            return new DummyObject(handle);
+        }
+    };
+
+    @Override
+    protected ChainedOptionsBuilder newOptionsBuilder() throws Exception {
+        return super.newOptionsBuilder().addProfiler("gc");
+    }
+
+    @Benchmark
+    public DummyObject plainNew() {
+        return new DummyObject();
+    }
+
+    @Benchmark
+    public DummyObject recyclerGetAndOrphan() {
+        return recycler.get();
+    }
+
+    @Benchmark
+    public DummyObject recyclerGetAndRecycle() {
+        DummyObject o = recycler.get();
+        o.recycle();
+        return o;
+    }
+
+    @State(Scope.Benchmark)
+    public static class ProducerConsumerState {
+        final ArrayBlockingQueue<DummyObject> queue = new ArrayBlockingQueue<DummyObject>(100);
+    }
+
+    // The allocation stats are the main thing interesting about this benchmark
+    @Benchmark
+    @Group("producerConsumer")
+    public void producer(ProducerConsumerState state, Control control) throws Exception {
+        ArrayBlockingQueue<DummyObject> queue = state.queue;
+        DummyObject object = recycler.get();
+        while (!control.stopMeasurement) {
+            if (queue.offer(object)) {
+                break;
+            }
+        }
+    }
+
+    @Benchmark
+    @Group("producerConsumer")
+    public void consumer(ProducerConsumerState state, Control control) throws Exception {
+        DummyObject object;
+        do {
+            object = state.queue.poll();
+            if (object != null) {
+                object.recycle();
+                return;
+            }
+        } while (!control.stopMeasurement);
+    }
+
+    @SuppressWarnings("unused")
+    private static final class DummyObject {
+        private final Recycler.Handle<DummyObject> handle;
+        private long l1;
+        private long l2;
+        private long l3;
+        private long l4;
+        private long l5;
+        private Object o1;
+        private Object o2;
+        private Object o3;
+        private Object o4;
+        private Object o5;
+
+        DummyObject() {
+            this(null);
+        }
+
+        DummyObject(Recycler.Handle<DummyObject> handle) {
+            this.handle = handle;
+        }
+
+        public void recycle() {
+            handle.recycle(this);
+        }
+    }
+}


Test patch:
diff --git a/common/src/test/java/io/netty/util/RecyclerTest.java b/common/src/test/java/io/netty/util/RecyclerTest.java
index 5d50e5981d1..76fdd465f32 100644
--- a/common/src/test/java/io/netty/util/RecyclerTest.java
+++ b/common/src/test/java/io/netty/util/RecyclerTest.java
@@ -28,6 +28,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -39,14 +40,17 @@
 public class RecyclerTest {
 
     private static Recycler<HandledObject> newRecycler(int maxCapacityPerThread) {
-        return newRecycler(maxCapacityPerThread, 2, 8, 2, 8);
+        return newRecycler(maxCapacityPerThread, 8, maxCapacityPerThread >> 1);
     }
 
     private static Recycler<HandledObject> newRecycler(int maxCapacityPerThread, int maxSharedCapacityFactor,
                                                        int ratio, int maxDelayedQueuesPerThread,
-                                                       int delayedQueueRatio) {
-        return new Recycler<HandledObject>(maxCapacityPerThread, maxSharedCapacityFactor, ratio,
-                maxDelayedQueuesPerThread, delayedQueueRatio) {
+                                                       int delayedQueueRatio, int chunkSize) {
+        return newRecycler(maxCapacityPerThread, ratio, chunkSize);
+    }
+
+    private static Recycler<HandledObject> newRecycler(int maxCapacityPerThread, int ratio, int chunkSize) {
+        return new Recycler<HandledObject>(maxCapacityPerThread, ratio, chunkSize) {
             @Override
             protected HandledObject newObject(
                     Recycler.Handle<HandledObject> handle) {
@@ -93,6 +97,11 @@ protected void finalize() throws Throwable {
         reference.getAndSet(null).recycle();
     }
 
+    @Test
+    public void verySmallRecycer() {
+        newRecycler(2, 0, 1).get();
+    }
+
     @Test
     public void testMultipleRecycle() {
         Recycler<HandledObject> recycler = newRecycler(1024);
@@ -187,7 +196,7 @@ public void run() {
             assertNotSame(a, b);
             IllegalStateException exception = exceptionStore.get();
             if (exception != null) {
-                assertEquals("recycled already", exception.getMessage());
+                assertThat(exception).hasMessageContaining("recycled already");
                 assertEquals(0, exception.getSuppressed().length);
             }
         } finally {
@@ -235,9 +244,7 @@ public void run() {
             HandledObject b = recycler.get();
             assertNotSame(a, b);
             IllegalStateException exception = exceptionStore.get();
-            if (exception != null) {
-                throw exception;
-            }
+            assertNotNull(exception); // Object got recycled twice, so at least one of the calls must throw.
         } finally {
             thread1.join(1000);
         }
@@ -265,7 +272,7 @@ public void testRecycleDisable() {
 
     @Test
     public void testRecycleDisableDrop() {
-        Recycler<HandledObject> recycler = newRecycler(1024, 2, 0, 2, 0);
+        Recycler<HandledObject> recycler = newRecycler(1024, 0, 16);
         HandledObject object = recycler.get();
         object.recycle();
         HandledObject object2 = recycler.get();
@@ -276,27 +283,6 @@ public void testRecycleDisableDrop() {
         object3.recycle();
     }
 
-    @Test
-    public void testRecycleDisableDelayedQueueDrop() throws Exception {
-        final Recycler<HandledObject> recycler = newRecycler(1024, 2, 1, 2, 0);
-        final HandledObject o = recycler.get();
-        final HandledObject o2 = recycler.get();
-        final HandledObject o3 = recycler.get();
-        final Thread thread = new Thread() {
-            @Override
-            public void run() {
-                o.recycle();
-                o2.recycle();
-                o3.recycle();
-            }
-        };
-        thread.start();
-        thread.join();
-        // In reverse order
-        assertSame(o3, recycler.get());
-        assertSame(o, recycler.get());
-    }
-
     /**
      * Test to make sure bug #2848 never happens again
      * https://github.com/netty/netty/issues/2848
@@ -322,14 +308,14 @@ private static void testMaxCapacity(int maxCapacity) {
             objects[i] = null;
         }
 
-        assertTrue(maxCapacity >= recycler.threadLocalCapacity(),
-                "The threadLocalCapacity (" + recycler.threadLocalCapacity() + ") must be <= maxCapacity ("
+        assertTrue(maxCapacity >= recycler.threadLocalSize(),
+                "The threadLocalSize (" + recycler.threadLocalSize() + ") must be <= maxCapacity ("
                 + maxCapacity + ") as we not pool all new handles internally");
     }
 
     @Test
     public void testRecycleAtDifferentThread() throws Exception {
-        final Recycler<HandledObject> recycler = newRecycler(256, 10, 2, 10, 2);
+        final Recycler<HandledObject> recycler = newRecycler(256, 2, 16);
         final HandledObject o = recycler.get();
         final HandledObject o2 = recycler.get();
 
@@ -387,7 +373,7 @@ public void run() {
     @Test
     public void testMaxCapacityWithRecycleAtDifferentThread() throws Exception {
         final int maxCapacity = 4; // Choose the number smaller than WeakOrderQueue.LINK_CAPACITY
-        final Recycler<HandledObject> recycler = newRecycler(maxCapacity);
+        final Recycler<HandledObject> recycler = newRecycler(maxCapacity, 4, 4);
 
         // Borrow 2 * maxCapacity objects.
         // Return the half from the same thread.
@@ -413,14 +399,12 @@ public void run() {
         thread.start();
         thread.join();
 
-        assertEquals(maxCapacity, recycler.threadLocalCapacity());
-        assertEquals(1, recycler.threadLocalSize());
+        assertEquals(maxCapacity * 3 / 4, recycler.threadLocalSize());
 
         for (int i = 0; i < array.length; i ++) {
             recycler.get();
         }
 
-        assertEquals(maxCapacity, recycler.threadLocalCapacity());
         assertEquals(0, recycler.threadLocalSize());
     }
 
@@ -429,7 +413,7 @@ public void testDiscardingExceedingElementsWithRecycleAtDifferentThread() throws
         final int maxCapacity = 32;
         final AtomicInteger instancesCount = new AtomicInteger(0);
 
-        final Recycler<HandledObject> recycler = new Recycler<HandledObject>(maxCapacity, 2) {
+        final Recycler<HandledObject> recycler = new Recycler<HandledObject>(maxCapacity) {
             @Override
             protected HandledObject newObject(Recycler.Handle<HandledObject> handle) {
                 instancesCount.incrementAndGet();