KAFKA-20302: Receive buffers allocated from MemoryPool may not be released if request is invalid by edoardocomar · Pull Request #21740 · apache/kafka

edoardocomar · 2026-03-13T02:15:58Z

If an exception is thrown within
kafka.network.Processor#processCompletedReceives close the receive
(return the buffer to the memory pool) if it has not been returned
already. Buffer may have been returned when successfully creating the
RequestChannel.Request if the api did not require DelayedAllocation

…eased if request is invalid Return the buffer to the pool if Execptions are caught within processCompletedReceives() taking care of not returning the buffer twice as it can be returned in the RequestChannel.Request constructor.

core/src/main/scala/kafka/network/SocketServer.scala

edoardocomar · 2026-03-16T15:33:07Z

thanks for your comments @akhileshchg

showuon · 2026-03-23T09:04:51Z

core/src/main/scala/kafka/network/SocketServer.scala

        // note that even though we got an exception, we can assume that receive.source is valid.
        // Issues with constructing a valid receive object were handled earlier
        case e: Throwable =>
+          if (header == null || req == null || header.apiKey.requiresDelayedAllocation) {


Question:

Normally, when will the buffer gets released if the header header.apiKey.requiresDelayedAllocation == true?

In this change, what happen if header.apiKey.requiresDelayedAllocation == false?

the buffer gets released if requiresDelayedAllocation is false
after being parsed inside the constructor of RequestChannel.Request
invoked on line 1027 of this PR version of SocketServer.scala
https://github.com/edoardocomar/kafka/blob/KAFKA-20302/core/src/main/scala/kafka/network/SocketServer.scala#L1027-L1028

but it only gets released there if (!header.apiKey.requiresDelayedAllocation)
see comment

//most request types are parsed entirely into objects at this point. for those we can release the underlying buffer. //some (like produce, or any time the schema contains fields of types BYTES or NULLABLE_BYTES) retain a reference //to the buffer. for those requests we cannot release the buffer early, but only when request processing is done.

https://github.com/edoardocomar/kafka/blob/KAFKA-20302/core/src/main/scala/kafka/network/RequestChannel.scala#L86-L105

in this PR if header.apiKey.requiresDelayedAllocation == false and in line 1054 we called close inconditionally, nothing wrong happens as NetworkReceive.close() is a no-op if invoked a 2nd time.

But checking for the condition in the catch makes the code more informative, IMHO

Thanks for the explanation. But could you point me where we will release the buffer when header.apiKey.requiresDelayedAllocation == true? It looks like we will close it when requestHandler completes the handling, is it correct? If so, then what will happen if the buffer is released in L1054 here, but then the request got processed with empty buffer?

Hi I see if the request has been successfully enqueued,
https://github.com/edoardocomar/kafka/blob/KAFKA-20302/core/src/main/scala/kafka/network/SocketServer.scala#L1040
then the buffer it will be eventually released at the end of KafkaRequesthandler.run
https://github.com/edoardocomar/kafka/blob/KAFKA-20302/core/src/main/scala/kafka/server/KafkaRequestHandler.scala#L171
but it will be accessed during the handling and it will be null

so we should not release in SocketServer if successfully enqueued into the requestChannel

Thanks @showuon I have restricted the release of the buffer in the catch to if (header == null || req == null)

as if successfully enqueued it will be released by the KafkaRequesthandler.

I checked with a manual test where I was throwing a random simulated IllegalStateException
at the end of handleChannelMuteEvent
and the pool balance was maintained.
I am unable to automate this test though - would need a mock KafkaChannel.handleChannelMuteEvent

i was wondering whether to move channel.channelMetadataRegistry.registerClientInformation
after the enqueuing of the request

but API_VERSIONS's buffer has been deallocated already as it's not a delayed schema (for now)

showuon · 2026-03-23T09:05:43Z

core/src/main/scala/kafka/network/SocketServer.scala

                // be sure to decrease connection count and drop any in-flight responses
                debug(s"Disconnecting expired channel: $channel : $header")
                close(channel.id)
+                receive.close() // return buffer to memory pool


Should we check header.apiKey.requiresDelayedAllocation here?

no there we should always close, as no RequestChannel.Request has been created there and no release of the buffer has occurred

showuon · 2026-03-23T09:09:09Z

...ration-tests/src/test/java/org/apache/kafka/clients/producer/SocketServerMemoryPoolTest.java

+//            }
+//        }
+//        return -1;
+//    }


I think we can keep the reason why we use reflection here. But I'm not sure if we need the example code in the comments here. How about we remove it?

I agree I wanted some feedback on this

alternatively we could make the memorypool accessor public with a comment // accessible for testing only ?

rewritten the test comment

edoardocomar · 2026-03-23T11:42:23Z

thanks @showuon for your comments. I believe I have addressed them. can you please re-review

…queued

edoardocomar added 3 commits March 13, 2026 01:47

braces

fa27612

single cleanup point

7e6704e

github-actions bot added triage PRs from the community core Kafka Broker small Small PRs labels Mar 13, 2026

integration test

333a290

github-actions bot added clients and removed small Small PRs labels Mar 13, 2026

akhileshchg reviewed Mar 13, 2026

View reviewed changes

core/src/main/scala/kafka/network/SocketServer.scala Show resolved Hide resolved

akhileshchg reviewed Mar 13, 2026

View reviewed changes

core/src/main/scala/kafka/network/SocketServer.scala Outdated Show resolved Hide resolved

akhileshchg reviewed Mar 13, 2026

View reviewed changes

core/src/main/scala/kafka/network/SocketServer.scala Show resolved Hide resolved

github-actions bot removed the triage PRs from the community label Mar 14, 2026

showuon self-assigned this Mar 17, 2026

showuon reviewed Mar 23, 2026

View reviewed changes

add comment following code review

e6e7fea

fix: do not release the buffer in SocketServer if request has been en…

d2dd7c6

…queued

Conversation

edoardocomar commented Mar 13, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

edoardocomar commented Mar 16, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar Mar 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar Mar 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar Mar 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edoardocomar commented Mar 23, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

edoardocomar commented Mar 13, 2026 •

edited by github-actions bot

Loading

edoardocomar Mar 23, 2026 •

edited

Loading

edoardocomar Mar 23, 2026 •

edited

Loading

edoardocomar Mar 24, 2026 •

edited

Loading