Netty Leak Detection firing sometimes on CI

[k-wall]

It is not clear why, but we have started to see reports of Netty Leak Detection. Its sporadic. It is not clear what has brought this out now.

logs_37921889964.zip

The first leak detected by the Maven plugin was after ResilienceIT.kafkaConsumerShouldTolerateKroxyliciousRestarting. The IT test before that was DeprecatedConfigurationIT (which also uses the Netty detector).

[ERROR] io.kroxylicious.proxy.ResilienceIT.kafkaConsumerShouldTolerateKroxyliciousRestarting(Topic) -- Time elapsed: 5.458 s <<< ERROR!
2025-04-30T18:44:51.7903806Z java.lang.IllegalStateException: Netty leaks: after [kafkaConsumerShouldTolerateKroxyliciousRestarting(Topic)] [ByteBuf]
2025-04-30T18:44:51.7905325Z    at io.github.nettyplus.leakdetector.junit.NettyLeakListener.assertZeroLeaks(NettyLeakListener.java:33)
2025-04-30T18:44:51.7906575Z    at io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension.afterEach(NettyLeakDetectorExtension.java:70)
2025-04-30T18:44:51.7907551Z    at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
2025-04-30T18:44:51.7908234Z    at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)

2025-04-30 18:44:41 ERROR io.netty.util.ResourceLeakDetector:337 - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:300)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
#2:
	io.netty.buffer.AdvancedLeakAwareByteBuf.readSlice(AdvancedLeakAwareByteBuf.java:119)
	io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder.opaqueFrame(KafkaRequestDecoder.java:201)
	io.kroxylicious.proxy.internal.codec.KafkaRequestDecoder.decodeHeaderAndBody(KafkaRequestDecoder.java:134)
	io.kroxylicious.proxy.internal.codec.KafkaMessageDecoder.decode(KafkaMessageDecoder.java:47)
	io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:530)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:469)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
#3:
	io.netty.buffer.AdvancedLeakAwareByteBuf.readInt(AdvancedLeakAwareByteBuf.java:437)
	io.kroxylicious.proxy.internal.codec.KafkaMessageDecoder.decode(KafkaMessageDecoder.java:36)
	io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:530)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:469)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
#4:
	Hint: 'requestDecoder' will handle the message from this point.
	io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:115)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:417)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
#5:
	Hint: 'DefaultChannelPipeline$HeadContext#0' will handle the message from this point.
	io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:115)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:417)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:410)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
	io.netty.channel.unix.PreferredDirectByteBufAllocator.ioBuffer(PreferredDirectByteBufAllocator.java:53)
	io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:120)
	io.netty.channel.epoll.EpollRecvByteAllocatorHandle.allocate(EpollRecvByteAllocatorHandle.java:75)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:784)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:501)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:399)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:1583)
: 6 leak records were discarded because the leak record count is targeted to 4. Use system property io.netty.leakDetection.targetRecords to increase the limit.

[k-wall]

I notice that in the case above, the issue is occuring after the ResilienceIT has run. I added shouldTolerateUpstreamGoingOffline so I wonder if that might be exposing the issue?

[k-wall]

I haven't been able to reproduce this issue locally. We are not seeing this issue regularly on CI either either. I think turning up io.netty.leakDetection.targetRecords would be a good idea so we have more information. We should gather more examples (add them to this issue)

From the Netty stack traces above, I think we know:

• The byte buf was read by IO - but it could be either upstream or downstream end.
• The RPC was following the opaque (undecoded) path
• The issues seems to have been triggered by DeprecatedConfigurationIT#testConsumerCanSurviveKroxyliciousRestart which is restarting the proxy (in-VM) whilst a consumer is connected and then verifying that the same consumer gets the next record.
• There are no filters in the test case.

[k-wall]

@SamBarker I'm not actually confident that the #2114 will resolve the issue. I couldn't make the story fit together. But let's see!