Fix netty buffer leak on short-circuit response #401

robobario · 2023-06-22T01:46:46Z

Type of change

Bugfix

Description

Netty ByteBuf uses explicit reference counting to control when they are released back to a pool. The buffer has to be released by someone before it is garbage collected or we have a leak.

A leak was introduced when we added the ability to forwardResponse while handling a Request. If the filter allocates a buffer using KrpcFilterContext#createByteBufferOutputStream, then the buffer is added to the DecodedFrame associated with the context. Then it is assumed that this frame will be read or written to the netty channel, so that netty can call release on it, which releases the buffers on the frame. In the short-circuit response case we create a new frame and the buffers on the old frame are left to be garbage collected.

Checklist

Please go through this checklist and make sure all applicable tasks have been done

Write tests
Make sure all tests pass
Make sure all Sonar-Lint warnings are addressed or are justifiably ignored.
Update documentation
Reference relevant issue(s) and close them after merging
For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

robobario · 2023-06-22T02:11:08Z

kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java

@@ -223,6 +223,7 @@ private void forwardShortCircuitResponse(ResponseHeaderData header, ApiMessage r
                    "Attempt to respond with ApiMessage of type " + ApiKeys.forId(response.apiKey()) + " but request is of type " + decodedFrame.apiKey());
        }
        DecodedResponseFrame<?> responseFrame = new DecodedResponseFrame<>(decodedFrame.apiVersion(), decodedFrame.correlationId(), header, response);
+        decodedFrame.transferBuffers(responseFrame);


Maybe these two lines would be better as decodedFrame.copy(header, response) but it's back to Generics hell since DecodedResponseFrame<B extends ApiMessage> has a generic type B for the ApiMessage, but here decodedFrame is DecodedFrame<?, ?> so we don't know the ApiMessage type it's dealing with. It could be done as in the other PR, remove the generic message type, and work in terms of ApiMessage with additional checks to ensure we only copy in the same concrete ApiMessage type.

Maybe best to get this fix in as is and look at refactoring the generics in another PR

I think we should get this in and refactor later.

robobario · 2023-06-22T02:24:32Z

I'm now wondering if we can also have a leak or unexpected behaviour if we use any netty buffers in the construction of a message that's used with sendRequest, I'm worried that the netty buffer could be released before the InternalRequestFrame is serialized into bytes if we forward the original frame before calling sendRequest (because all the netty buffers are released when the original frame is handled).

I tried it experimentally and it seems to work okay, maybe because in the testing environment we don't have anything else trying to allocate buffers that could get allocated that released memory.

@franz1981 am I right that that could be a problem? In miniature it would look like

        final ByteBuf buffer = channelContext.alloc().ioBuffer(initialCapacity);
        ByteBuffer nioBuffer = byteBuf.nioBuffer(byteBuf.writerIndex(), byteBuf.writableBytes())
        buffer.release()
        // other code continues to use nioBuffer

I imagine this mean that something else could be allocated the same memory after the release. I've confirmed this with a little playing around like:

        PooledByteBufAllocator pooledByteBufAllocator = new PooledByteBufAllocator();
        ByteBuf byteBuf = pooledByteBufAllocator.ioBuffer(100);
        ByteBuffer byteBuffer = byteBuf.nioBuffer(byteBuf.writerIndex(), byteBuf.writableBytes());
        byteBuf.release();
        byteBuffer.mark();
        for (int i = 0; i < 100; i++) {
            byteBuffer.put((byte)5);
        }
        byteBuffer.rewind();
        ByteBuf byteBuf1 = pooledByteBufAllocator.ioBuffer(100);
        ByteBuffer byteBuffer1 = byteBuf1.nioBuffer(byteBuf1.writerIndex(), byteBuf1.writableBytes());
        System.out.println(byteBuffer1.get(0));

        byteBuffer.mark();
        for (int i = 0; i < 100; i++) {
            byteBuffer.put((byte)2);
        }
        byteBuffer.rewind();
        System.out.println(byteBuffer1.get(0));

Which prints 5 then 2, so I think I'm happy that it would be a bad thing.

SamBarker

Thanks for looking into this @robobario.

Lets fix this now and clean things up later.

SamBarker · 2023-06-22T22:39:23Z

integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java

+
+        final LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+        final Configuration config = ctx.getConfiguration();
+        appender = (NettyLeakLogAppender) config.getAppenders().get("NettyLeakLogAppender");


Urgh! This is both ingenious and nasty.

I think it still suffers from the fact the leak detector can only work on GC cycles, it detects things being freed with a positive reference count. Maybe after each should also request a GC cycle.

edit: I see there is an attempt to force GC in the test itself.

Yeahp a bit painful, it can possibly cause a non-deterministic failure in an unrelated test in KrpcFilterIT, but the leak trace should point you in the right direction. Better than a silent leak :(

integrationtests/src/test/java/io/kroxylicious/proxy/KrpcFilterIT.java

SamBarker · 2023-06-22T22:53:13Z

kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java

@@ -223,6 +223,7 @@ private void forwardShortCircuitResponse(ResponseHeaderData header, ApiMessage r
                    "Attempt to respond with ApiMessage of type " + ApiKeys.forId(response.apiKey()) + " but request is of type " + decodedFrame.apiKey());
        }
        DecodedResponseFrame<?> responseFrame = new DecodedResponseFrame<>(decodedFrame.apiVersion(), decodedFrame.correlationId(), header, response);
+        decodedFrame.transferBuffers(responseFrame);


I think we should get this in and refactor later.

SamBarker · 2023-06-22T22:54:46Z

kroxylicious/src/main/java/io/kroxylicious/proxy/internal/DefaultFilterContext.java

@@ -223,6 +223,7 @@ private void forwardShortCircuitResponse(ResponseHeaderData header, ApiMessage r
                    "Attempt to respond with ApiMessage of type " + ApiKeys.forId(response.apiKey()) + " but request is of type " + decodedFrame.apiKey());
        }
        DecodedResponseFrame<?> responseFrame = new DecodedResponseFrame<>(decodedFrame.apiVersion(), decodedFrame.correlationId(), header, response);
+        decodedFrame.transferBuffers(responseFrame);


I think you mentioned elsewhere why its this way round but it feels odd that we are transferring a responseFrame to a request frame. Naively I would expect us to keep the request frame around until we had finished with the response not the other way around.

It's the other way around, I've renamed the method to decodedFrame.transferBuffersTo(responseFrame)

I guess part of what confused me so the pushing of buffers rather than pulling but I guess it has to be that way round for access

SamBarker · 2023-06-22T22:58:28Z

integrationtests/src/test/java/io/kroxylicious/proxy/NettyLeakLogAppender.java

+import io.netty.util.ResourceLeakDetector;
+
+@Plugin(name = "NettyLeakLogAppender", category = "Core", elementType = "appender", printObject = true)
+public class NettyLeakLogAppender extends AbstractAppender {


Thinking out loud I wonder if its worth adding this to the junit extension so we can apply it to all the integration tests?

Its the sort of thing I think would be good to support filter authors in verifying as well.

is the junit5 extension that right place? the extension doesn't actually know anything about netty.

Yeah I did wonder about that. Maybe a separate extension in Kroxylicious test tools

franz1981 · 2023-06-23T13:46:56Z

Related the leak, yep, if nio buffer is stolen should be used temporarily and I have to verify if retaining the originating ByteBuf would help (because the NIO buffer is a view which assume the owner won't be shared, so probably not).
The reason why I have done it at

kroxylicious/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/ByteBufOutputStream.java

Line 33 in 4b7a824

    
           this.nioBuffer = byteBuf.nioBuffer(byteBuf.writerIndex(), byteBuf.writableBytes());

was to bridge our uses to what Kafka offers in term of API and avoid a copy (and allocation of temporary NIO buffer, which, given the length of Kafka frames can be impactful).
Said that, such uses should be temporarily and rely instead of ownership control of Netty buffers instead, when it makes sense (retain is a costy operation and calls should make evident when the ownership is fully transferred or shared, avoiding the retain in the former case)

Netty ByteBuf uses explicit reference counting to control when they are released back to a pool. The buffer has to be released by someone before it is garbage collected or we have a leak. A leak was introduced when we added the ability to `forwardResponse` while handling a Request. If the filter allocates a buffer using KrpcFilterContext#createByteBufferOutputStream, then the buffer is added to the DecodedFrame associated with the context. Then it is assumed that this frame will be read or written to the netty channel, so that netty can call release on it, which releases the buffers on the frame. In the short-circuit response case we create a new frame and the buffers on the old frame are left to be garbage collected.

A leak was introduced when we added the ability to `forwardResponse` while handling a Request. If the filter allocates a buffer using KrpcFilterContext#createByteBufferOutputStream, then the buffer is added to the DecodedFrame associated with the context. Then it is assumed that this frame will be read or written to the netty channel, so that netty can call release on it, which releases the buffers on the frame. In the short-circuit response case we create a new frame and the buffers on the old frame are left to be garbage collected. With this solution we transfer the buffers to the new frame that is handed to netty.

robobario force-pushed the fix-buffer-leak branch from caaec67 to 992566b Compare June 22, 2023 02:03

robobario commented Jun 22, 2023

View reviewed changes

robobario marked this pull request as ready for review June 22, 2023 02:13

SamBarker approved these changes Jun 22, 2023

View reviewed changes

SamBarker reviewed Jun 22, 2023

View reviewed changes

robobario force-pushed the fix-buffer-leak branch 2 times, most recently from 5bf3779 to 202d7b7 Compare June 26, 2023 21:26

robobario merged commit 342738e into kroxylicious:main Jun 26, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix netty buffer leak on short-circuit response #401

Fix netty buffer leak on short-circuit response #401

robobario commented Jun 22, 2023 •

edited

Loading

robobario Jun 22, 2023 •

edited

Loading

robobario Jun 22, 2023

SamBarker Jun 22, 2023

robobario commented Jun 22, 2023 •

edited

Loading

SamBarker left a comment •

edited

Loading

SamBarker Jun 22, 2023

robobario Jun 26, 2023

SamBarker Jun 22, 2023

SamBarker Jun 22, 2023

robobario Jun 26, 2023

SamBarker Jun 27, 2023

SamBarker Jun 22, 2023

k-wall Jun 23, 2023 •

edited

Loading

SamBarker Jun 23, 2023

franz1981 commented Jun 23, 2023 •

edited

Loading

Fix netty buffer leak on short-circuit response #401

Fix netty buffer leak on short-circuit response #401

Conversation

robobario commented Jun 22, 2023 • edited Loading

Type of change

Description

Checklist

robobario Jun 22, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robobario commented Jun 22, 2023 • edited Loading

SamBarker left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

k-wall Jun 23, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

franz1981 commented Jun 23, 2023 • edited Loading

robobario commented Jun 22, 2023 •

edited

Loading

robobario Jun 22, 2023 •

edited

Loading

robobario commented Jun 22, 2023 •

edited

Loading

SamBarker left a comment •

edited

Loading

k-wall Jun 23, 2023 •

edited

Loading

franz1981 commented Jun 23, 2023 •

edited

Loading