This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 82712644a65 KAFKA-20451 Move RequestChannel Responses to server module
(#22517)
82712644a65 is described below
commit 82712644a65716c270618fc8a035847788120e44
Author: Mickael Maison <[email protected]>
AuthorDate: Fri Jul 10 16:36:36 2026 +0200
KAFKA-20451 Move RequestChannel Responses to server module (#22517)
This pull request migrates the `Response` class hierarchy from Scala to
native Java records within the org.apache.kafka.network package to
improve interoperability.
Reviewers: TaiJuWu <[email protected]>, Ken Huang <[email protected]>,
Chia-Ping Tsai <[email protected]>
---
.../main/scala/kafka/network/RequestChannel.scala | 56 ++++------------------
.../main/scala/kafka/network/SocketServer.scala | 23 ++++-----
.../kafka/network/RequestConvertToJsonTest.scala | 10 ++--
.../unit/kafka/network/SocketServerTest.scala | 14 +++---
.../scala/unit/kafka/server/KafkaApisTest.scala | 11 ++---
.../kafka/network/CloseConnectionResponse.java | 25 ++++++++++
.../kafka/network/EndThrottlingResponse.java | 25 ++++++++++
.../org/apache/kafka/network/NoOpResponse.java | 25 ++++++++++
.../java/org/apache/kafka/network/Response.java | 30 ++++++++++++
.../org/apache/kafka/network/SendResponse.java | 34 +++++++++++++
.../kafka/network/StartThrottlingResponse.java | 25 ++++++++++
11 files changed, 199 insertions(+), 79 deletions(-)
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala
b/core/src/main/scala/kafka/network/RequestChannel.scala
index ce776a1ee9f..3c8f0d140d2 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -18,63 +18,23 @@
package kafka.network
import java.util.concurrent._
-import com.fasterxml.jackson.databind.JsonNode
import kafka.utils.Logging
-import org.apache.kafka.common.network.Send
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests._
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.metrics.internals.MetricsUtils
-import org.apache.kafka.network.{BaseRequest, CallbackRequest, Request,
ShutdownRequest, WakeupRequest}
+import org.apache.kafka.network.{BaseRequest, CallbackRequest,
CloseConnectionResponse, EndThrottlingResponse, NoOpResponse, Request,
Response, SendResponse, ShutdownRequest, StartThrottlingResponse, WakeupRequest}
import org.apache.kafka.network.metrics.RequestChannelMetrics
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import java.util.OptionalLong
import scala.jdk.CollectionConverters._
-import scala.jdk.OptionConverters._
object RequestChannel extends Logging {
private val RequestQueueSizeMetric = "RequestQueueSize"
private val ResponseQueueSizeMetric = "ResponseQueueSize"
val ProcessorMetricTag = "processor"
-
- sealed abstract class Response(val request: Request) {
-
- def processor: Int = request.processor
-
- def responseLog: Option[JsonNode] = None
- }
-
- /** responseLogValue should only be defined if request logging is enabled */
- class SendResponse(request: Request,
- val responseSend: Send,
- val responseLogValue: Option[JsonNode]) extends
Response(request) {
- override def responseLog: Option[JsonNode] = responseLogValue
-
- override def toString: String =
- s"Response(type=Send, request=$request, send=$responseSend,
asString=$responseLogValue)"
- }
-
- class NoOpResponse(request: Request) extends Response(request) {
- override def toString: String =
- s"Response(type=NoOp, request=$request)"
- }
-
- class CloseConnectionResponse(request: Request) extends Response(request) {
- override def toString: String =
- s"Response(type=CloseConnection, request=$request)"
- }
-
- class StartThrottlingResponse(request: Request) extends Response(request) {
- override def toString: String =
- s"Response(type=StartThrottling, request=$request)"
- }
-
- class EndThrottlingResponse(request: Request) extends Response(request) {
- override def toString: String =
- s"Response(type=EndThrottling, request=$request)"
- }
}
class RequestChannel(val queueSize: Int,
@@ -125,7 +85,7 @@ class RequestChannel(val queueSize: Int,
// This case is used when the request handler has encountered an error,
but the client
// does not expect a response (e.g. when produce request has acks set to 0)
updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
- sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ sendResponse(new CloseConnectionResponse(request))
}
def sendResponse(
@@ -133,19 +93,19 @@ class RequestChannel(val queueSize: Int,
response: AbstractResponse
): Unit = {
updateErrorMetrics(request.header.apiKey, response.errorCounts.asScala)
- sendResponse(new RequestChannel.SendResponse(
+ sendResponse(new SendResponse(
request,
request.buildResponseSend(response),
- request.responseNode(response).toScala
+ request.responseNode(response)
))
}
def sendNoOpResponse(request: Request): Unit = {
- sendResponse(new RequestChannel.NoOpResponse(request))
+ sendResponse(new NoOpResponse(request))
}
def startThrottling(request: Request): Unit = {
- sendResponse(new RequestChannel.StartThrottlingResponse(request))
+ sendResponse(new StartThrottlingResponse(request))
}
def endThrottling(request: Request): Unit = {
@@ -153,7 +113,7 @@ class RequestChannel(val queueSize: Int,
}
/** Send a response back to the socket server to be sent over the network */
- private[network] def sendResponse(response: RequestChannel.Response): Unit =
{
+ private[network] def sendResponse(response: Response): Unit = {
if (isTraceEnabled) {
val requestHeader = response.request.headerForLoggingOrThrottling()
val message = response match {
@@ -186,7 +146,7 @@ class RequestChannel(val queueSize: Int,
case _: StartThrottlingResponse | _: EndThrottlingResponse => ()
}
- val processor = processors.get(response.processor)
+ val processor = processors.get(response.request.processor)
// The processor may be null if it was shutdown. In this case, the
connections
// are closed, so the response is dropped.
if (processor != null) {
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala
b/core/src/main/scala/kafka/network/SocketServer.scala
index 50fe55427bf..291e689d2bd 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -26,7 +26,7 @@ import java.util.Optional
import java.util.concurrent._
import java.util.concurrent.atomic._
import kafka.network.Processor._
-import kafka.network.RequestChannel.{CloseConnectionResponse,
EndThrottlingResponse, NoOpResponse, SendResponse, StartThrottlingResponse}
+import org.apache.kafka.network.{CloseConnectionResponse,
EndThrottlingResponse, NoOpResponse, Response, SendResponse,
StartThrottlingResponse}
import kafka.server.{BrokerReconfigurable, KafkaConfig}
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import kafka.utils._
@@ -58,7 +58,6 @@ import org.slf4j.event.Level
import scala.collection._
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
-import scala.jdk.OptionConverters._
import scala.util.control.ControlThrowable
/**
@@ -829,8 +828,8 @@ private[kafka] class Processor(
val thread: KafkaThread = KafkaThread.nonDaemon(threadName, this)
private val newConnections = new
ArrayBlockingQueue[SocketChannel](connectionQueueSize)
- private val inflightResponses = mutable.Map[String,
RequestChannel.Response]()
- private val responseQueue = new
LinkedBlockingDeque[RequestChannel.Response]()
+ private val inflightResponses = mutable.Map[String, Response]()
+ private val responseQueue = new LinkedBlockingDeque[Response]()
private[kafka] val metricTags = mutable.LinkedHashMap(
ListenerMetricTag -> listenerName.value,
@@ -935,7 +934,7 @@ private[kafka] class Processor(
}
private def processNewResponses(): Unit = {
- var currentResponse: RequestChannel.Response = null
+ var currentResponse: Response = null
while ({currentResponse = dequeueResponse(); currentResponse != null}) {
val channelId = currentResponse.request.context.connectionId
try {
@@ -964,8 +963,6 @@ private[kafka] class Processor(
// the client.
handleChannelMuteEvent(channelId, ChannelMuteEvent.THROTTLE_ENDED)
tryUnmuteChannel(channelId)
- case _ =>
- throw new IllegalArgumentException(s"Unknown response type:
${currentResponse.getClass}")
}
} catch {
case e: Throwable =>
@@ -975,13 +972,13 @@ private[kafka] class Processor(
}
// `protected` for test usage
- protected[network] def sendResponse(response: RequestChannel.Response,
responseSend: Send): Unit = {
+ protected[network] def sendResponse(response: Response, responseSend: Send):
Unit = {
val connectionId = response.request.context.connectionId
trace(s"Socket server received response to send to $connectionId,
registering for write and sending data: $response")
// `channel` can be None if the connection was closed remotely or if
selector closed it for being idle for too long
if (channel(connectionId).isEmpty) {
warn(s"Attempting to send response via channel for which there is no
open connection, connection id $connectionId")
- response.request.updateRequestMetrics(0L, response.responseLog.toJava)
+ response.request.updateRequestMetrics(0L, response.responseLog)
}
// Invoke send for closingChannel as well so that the send is failed and
the channel closed properly and
// removed from the Selector after discarding any pending staged receives.
@@ -1083,10 +1080,10 @@ private[kafka] class Processor(
selector.clearCompletedSends()
}
- private def updateRequestMetrics(response: RequestChannel.Response): Unit = {
+ private def updateRequestMetrics(response: Response): Unit = {
val request = response.request
val networkThreadTimeNanos =
openOrClosingChannel(request.context.connectionId).fold(0L)(_.getAndResetNetworkThreadTimeNanos())
- request.updateRequestMetrics(networkThreadTimeNanos,
response.responseLog.toJava)
+ request.updateRequestMetrics(networkThreadTimeNanos, response.responseLog)
}
private def processDisconnected(): Unit = {
@@ -1205,12 +1202,12 @@ private[kafka] class Processor(
connId
}
- private[network] def enqueueResponse(response: RequestChannel.Response):
Unit = {
+ private[network] def enqueueResponse(response: Response): Unit = {
responseQueue.put(response)
wakeup()
}
- private def dequeueResponse(): RequestChannel.Response = {
+ private def dequeueResponse(): Response = {
val response = responseQueue.poll()
if (response != null)
response.request.responseDequeueTimeNanos(Time.SYSTEM.nanoseconds)
diff --git
a/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala
b/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala
index c8dc9e8e729..621f59194f9 100644
--- a/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala
+++ b/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala
@@ -20,19 +20,19 @@ package kafka.network
import java.net.InetAddress
import java.nio.ByteBuffer
import com.fasterxml.jackson.databind.node.{BooleanNode, DoubleNode,
JsonNodeFactory, LongNode, NullNode, ObjectNode, TextNode}
+import java.util.Optional
import org.apache.kafka.common.memory.MemoryPool
import org.apache.kafka.common.message._
import org.apache.kafka.common.network.{ClientInformation, ListenerName,
NetworkSend}
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.requests._
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.network.{Request, RequestConvertToJson}
+import org.apache.kafka.network.{Request, RequestConvertToJson, SendResponse}
import org.apache.kafka.network.metrics.RequestChannelMetrics
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import org.mockito.Mockito.mock
-import scala.jdk.OptionConverters.RichOption
class RequestConvertToJsonTest {
@@ -71,7 +71,7 @@ class RequestConvertToJsonTest {
val req = request(alterIsrRequest)
val send = new NetworkSend(req.context.connectionId,
alterIsrRequest.toSend(req.header))
val headerLog = RequestConvertToJson.requestHeaderNode(req.header)
- val res = new RequestChannel.SendResponse(req, send, Some(headerLog))
+ val res = new SendResponse(req, send, Optional.of(headerLog))
val totalTimeMs = 1
val requestQueueTimeMs = 2
@@ -84,7 +84,7 @@ class RequestConvertToJsonTest {
val messageConversionsTimeMs = 9
val expectedNode = RequestConvertToJson.requestDesc(req.header,
req.requestLog, req.isForwarded).asInstanceOf[ObjectNode]
- expectedNode.set("response",
res.responseLog.getOrElse(NullNode.getInstance()))
+ expectedNode.set("response",
res.responseLog.orElse(NullNode.getInstance()))
expectedNode.set("connection", new TextNode(req.context.connectionId))
expectedNode.set("totalTimeMs", new DoubleNode(totalTimeMs))
expectedNode.set("requestQueueTimeMs", new DoubleNode(requestQueueTimeMs))
@@ -100,7 +100,7 @@ class RequestConvertToJsonTest {
expectedNode.set("temporaryMemoryBytes", new
LongNode(temporaryMemoryBytes))
expectedNode.set("messageConversionsTime", new
DoubleNode(messageConversionsTimeMs))
- val actualNode = RequestConvertToJson.requestDescMetrics(req.header,
req.requestLog, res.responseLog.toJava, req.context, req.session,
req.isForwarded,
+ val actualNode = RequestConvertToJson.requestDescMetrics(req.header,
req.requestLog, res.responseLog, req.context, req.session, req.isForwarded,
totalTimeMs, requestQueueTimeMs, apiLocalTimeMs, apiRemoteTimeMs,
apiThrottleTimeMs, responseQueueTimeMs,
responseSendTimeMs, temporaryMemoryBytes,
messageConversionsTimeMs).asInstanceOf[ObjectNode]
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index 4e0de77c247..275b245532d 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -36,7 +36,7 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal,
SecurityProtocol}
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.utils._
import org.apache.kafka.common.utils.internals.{AppInfoParser, LogContext}
-import org.apache.kafka.network.{CallbackRequest, Request,
RequestConvertToJson, ShutdownRequest, SocketServerConfigs, WakeupRequest}
+import org.apache.kafka.network.{CallbackRequest, NoOpResponse, Request,
RequestConvertToJson, Response, SendResponse, ShutdownRequest,
SocketServerConfigs, WakeupRequest}
import org.apache.kafka.security.CredentialProvider
import org.apache.kafka.server.{ApiVersionManager, SimpleApiVersionManager}
import org.apache.kafka.server.common.{FinalizedFeatures, MetadataVersion}
@@ -60,7 +60,7 @@ import java.security.cert.X509Certificate
import java.util
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent._
-import java.util.{Properties, Random}
+import java.util.{Optional, Properties, Random}
import javax.net.ssl._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -164,7 +164,7 @@ class SocketServerTest {
val byteBuffer =
request.body(classOf[AbstractRequest]).serializeWithHeader(request.header)
val send = new NetworkSend(request.context.connectionId,
ByteBufferSend.sizePrefixed(byteBuffer))
val headerLog = RequestConvertToJson.requestHeaderNode(request.header)
- channel.sendResponse(new RequestChannel.SendResponse(request, send,
Some(headerLog)))
+ channel.sendResponse(new SendResponse(request, send,
Optional.of(headerLog)))
}
def processRequestNoOpResponse(channel: RequestChannel, request: Request):
Unit = {
@@ -652,9 +652,9 @@ class SocketServerTest {
val headerLog = RequestConvertToJson.requestHeaderNode(request.header)
val response =
if (!noOpResponse)
- new RequestChannel.SendResponse(request, send, Some(headerLog))
+ new SendResponse(request, send, Optional.of(headerLog))
else
- new RequestChannel.NoOpResponse(request)
+ new NoOpResponse(request)
server.dataPlaneRequestChannel.sendResponse(response)
// Quota manager would call notifyThrottlingDone() on throttling
completion. Simulate it if throttlingInProgress is
@@ -1104,7 +1104,7 @@ class SocketServerTest {
val send = new NetworkSend(request.context.connectionId,
ByteBufferSend.sizePrefixed(ByteBuffer.allocate(responseBufferSize)))
val headerLog = new ObjectNode(JsonNodeFactory.instance)
headerLog.set("response", new TextNode("someResponse"))
- channel.sendResponse(new RequestChannel.SendResponse(request, send,
Some(headerLog)))
+ channel.sendResponse(new SendResponse(request, send,
Optional.of(headerLog)))
TestUtils.waitUntilTrue(() => totalTimeHistCount() ==
expectedTotalTimeCount,
s"request metrics not updated, expected: $expectedTotalTimeCount,
actual: ${totalTimeHistCount()}")
@@ -2083,7 +2083,7 @@ class SocketServerTest {
this.conn = Some(conn)
}
- override protected[network] def sendResponse(response:
RequestChannel.Response, responseSend: Send): Unit = {
+ override protected[network] def sendResponse(response: Response,
responseSend: Send): Unit = {
this.conn.foreach(_.close())
super.sendResponse(response, responseSend)
}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 97b458e27b3..a0079dd99eb 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -88,7 +88,7 @@ import org.apache.kafka.coordinator.share.{ShareCoordinator,
ShareCoordinatorTes
import org.apache.kafka.coordinator.transaction.{InitProducerIdResult,
TransactionLogConfig}
import org.apache.kafka.image.{MetadataDelta, MetadataImage,
MetadataProvenance}
import org.apache.kafka.metadata.{ConfigRepository, KRaftMetadataCache,
MetadataCache, MetadataCacheFixtures, MockConfigRepository}
-import org.apache.kafka.network.{Request, Session}
+import org.apache.kafka.network.{Request, SendResponse, Session}
import org.apache.kafka.network.metrics.{RequestChannelMetrics, RequestMetrics}
import org.apache.kafka.raft.{KRaftConfigs, QuorumConfig}
import org.apache.kafka.security.authorizer.AclEntry
@@ -127,7 +127,6 @@ import java.util.function.Consumer
import java.util.{Comparator, Optional, OptionalInt, OptionalLong, Properties}
import scala.collection.{Map, Seq, mutable}
import scala.jdk.CollectionConverters._
-import scala.jdk.OptionConverters._
class KafkaApisTest extends Logging {
private val requestChannel: RequestChannel = mock(classOf[RequestChannel])
@@ -10412,13 +10411,13 @@ class KafkaApisTest extends Logging {
request.context.header.apiVersion
)
- // Create the RequestChannel.Response that is created when sendResponse is
called in order to update the metrics.
- val sendResponse = new RequestChannel.SendResponse(
+ // Create the Response that is created when sendResponse is called in
order to update the metrics.
+ val sendResponse = new SendResponse(
request,
request.buildResponseSend(response),
- request.responseNode(response).toScala
+ request.responseNode(response)
)
- request.updateRequestMetrics(time.milliseconds(),
sendResponse.responseLog.toJava)
+ request.updateRequestMetrics(time.milliseconds(), sendResponse.responseLog)
AbstractResponse.parseResponse(
request.context.header.apiKey,
diff --git
a/server/src/main/java/org/apache/kafka/network/CloseConnectionResponse.java
b/server/src/main/java/org/apache/kafka/network/CloseConnectionResponse.java
new file mode 100644
index 00000000000..73fa00ce854
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/CloseConnectionResponse.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+public record CloseConnectionResponse(Request request) implements Response {
+
+ @Override
+ public String toString() {
+ return "Response(type=CloseConnection, request=" + request + ")";
+ }
+}
diff --git
a/server/src/main/java/org/apache/kafka/network/EndThrottlingResponse.java
b/server/src/main/java/org/apache/kafka/network/EndThrottlingResponse.java
new file mode 100644
index 00000000000..c9fcacb7f6d
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/EndThrottlingResponse.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+public record EndThrottlingResponse(Request request) implements Response {
+
+ @Override
+ public String toString() {
+ return "Response(type=EndThrottling, request=" + request + ")";
+ }
+}
diff --git a/server/src/main/java/org/apache/kafka/network/NoOpResponse.java
b/server/src/main/java/org/apache/kafka/network/NoOpResponse.java
new file mode 100644
index 00000000000..8c7ad667d8e
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/NoOpResponse.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+public record NoOpResponse(Request request) implements Response {
+
+ @Override
+ public String toString() {
+ return "Response(type=NoOp, request=" + request + ")";
+ }
+}
diff --git a/server/src/main/java/org/apache/kafka/network/Response.java
b/server/src/main/java/org/apache/kafka/network/Response.java
new file mode 100644
index 00000000000..eb51c0fcc07
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/Response.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.Optional;
+
+public sealed interface Response permits SendResponse, NoOpResponse,
CloseConnectionResponse, StartThrottlingResponse, EndThrottlingResponse {
+
+ Request request();
+
+ default Optional<JsonNode> responseLog() {
+ return Optional.empty();
+ }
+}
diff --git a/server/src/main/java/org/apache/kafka/network/SendResponse.java
b/server/src/main/java/org/apache/kafka/network/SendResponse.java
new file mode 100644
index 00000000000..2d991700514
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/SendResponse.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+import org.apache.kafka.common.network.Send;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.Optional;
+
+/**
+ * @param responseLog should only be defined if request logging is enabled
+ */
+public record SendResponse(Request request, Send responseSend,
Optional<JsonNode> responseLog) implements Response {
+
+ @Override
+ public String toString() {
+ return "Response(type=Send, request=" + request + ", send=" +
responseSend + ", asString=" + responseLog + ")";
+ }
+}
diff --git
a/server/src/main/java/org/apache/kafka/network/StartThrottlingResponse.java
b/server/src/main/java/org/apache/kafka/network/StartThrottlingResponse.java
new file mode 100644
index 00000000000..016745545a4
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/network/StartThrottlingResponse.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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
+ *
+ * http://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 org.apache.kafka.network;
+
+public record StartThrottlingResponse(Request request) implements Response {
+
+ @Override
+ public String toString() {
+ return "Response(type=StartThrottling, request=" + request + ")";
+ }
+}
