fix(network): fix potential ByteBuf LEAK in fetch

[superhx]

No description provided.

[Copilot]

Pull Request Overview

This PR addresses potential ByteBuf memory leaks in the fetch request handling by ensuring proper resource cleanup when requests fail or connections are closed.

Key changes:

• Adds resource release logic for fetch responses when fetch session epoch validation fails
• Implements a removeChannelContext method to properly clean up resources when connections are closed
• Moves the MAX_INFLIGHT_REQUESTS_PER_CONNECTION constant from config to SocketServer object for better code organization

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.

File	Description
server/src/main/java/org/apache/kafka/network/SocketServerConfigs.java	Removes the MAX_INFLIGHT_REQUESTS_PER_CONNECTION constant from config class
core/src/main/scala/kafka/server/FetchSession.scala	Adds resource cleanup for pooled fetch records when session epoch validation fails
core/src/main/scala/kafka/network/SocketServer.scala	Moves constant definition, adds removeChannelContext method for proper resource cleanup, and updates references to use the new constant location

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Similarly, the resource release in the fetch session error handling lacks exception handling. If r.release() throws an exception for one record, subsequent records won't be released, leading to potential resource leaks. Consider adding try-catch around the release operation to ensure all resources are attempted to be released.

Suggested change

	r.release()
	try {
	r.release()
	} catch {
	case ex: Throwable =>
	info(s"Exception while releasing PooledResource in fetch session error handling: ${ex.getMessage}", ex)
	}

[Copilot]

The semicolon at the end of this line is inconsistent with Scala conventions. Scala doesn't require semicolons at the end of statements, and they are typically omitted unless multiple statements are on the same line.

Suggested change

	val MaxInflightRequestsPerConnection = 64;
	val MaxInflightRequestsPerConnection = 64

[Copilot]

In the removeChannelContext method, both clear() calls at lines 1456 and 1464 are executed inside a synchronized block on channelContext. However, the forEach iteration at lines 1457-1463 is also inside this synchronized block, which could potentially cause a deadlock or blocking issue if the release() method on responseSend acquires other locks or performs time-consuming operations. Consider releasing resources outside the synchronized block or ensure that the release() method doesn't acquire locks that could cause contention.

Suggested change

	channelContext.synchronized {
	channelContext.nextCorrelationId.clear()
	channelContext.responses.forEach((_, response) => {
	response match {
	case sendResponse: SendResponse =>
	sendResponse.responseSend.release()
	case _ =>
	}
	})
	channelContext.responses.clear()
	}
	// Collect SendResponse objects to release outside the synchronized block
	val sendResponses = new ArrayBuffer[SendResponse]()
	channelContext.synchronized {
	channelContext.nextCorrelationId.clear()
	channelContext.responses.forEach((_, response) => {
	response match {
	case sendResponse: SendResponse =>
	sendResponses += sendResponse
	case _ =>
	}
	})
	channelContext.responses.clear()
	}
	// Release resources outside the synchronized block
	sendResponses.foreach(_.responseSend.release())

[Copilot]

The error handling in removeChannelContext doesn't account for potential exceptions that could be thrown during the release() call. If responseSend.release() throws an exception, it could leave the channelContext in an inconsistent state with some responses released and others not. Consider wrapping the release operation in a try-catch block to ensure all resources are attempted to be released even if some fail.

Suggested change

	sendResponse.responseSend.release()
	try {
	sendResponse.responseSend.release()
	} catch {
	case e: Throwable =>
	error(s"Error releasing responseSend for connection $connectionId", e)
	}