KAFKA-17601: Inter-broker connections do not expose their clientSoftwareName and clientSoftwareVersion tags

Signed-off-by: PoAn Yang <[email protected]>

Author: FrankYang0529

Diff for: clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java

@@ -109,7 +109,11 @@ public RequestContext(RequestHeader header,
     }
 
     public RequestAndSize parseRequest(ByteBuffer buffer) {
-        if (isUnsupportedApiVersionsRequest()) {
+        return staticParseRequest(header, buffer, connectionId, listenerName, principal);
+    }
+
+    public static RequestAndSize staticParseRequest(RequestHeader header, ByteBuffer buffer, String connectionId, ListenerName listenerName, KafkaPrincipal principal) {
+        if (isUnsupportedApiVersionsRequest(header)) {
             // Unsupported ApiVersion requests are treated as v0 requests and are not parsed
             ApiVersionsRequest apiVersionsRequest = new ApiVersionsRequest(new ApiVersionsRequestData(), (short) 0, header.apiVersion());
             return new RequestAndSize(apiVersionsRequest, 0);
@@ -150,13 +154,13 @@ public ByteBuffer buildResponseEnvelopePayload(AbstractResponse body) {
         return body.serializeWithHeader(header.toResponseHeader(), apiVersion());
     }
 
-    private boolean isUnsupportedApiVersionsRequest() {
+    private static boolean isUnsupportedApiVersionsRequest(RequestHeader header) {
         return header.apiKey() == API_VERSIONS && !API_VERSIONS.isVersionSupported(header.apiVersion());
     }
 
     public short apiVersion() {
         // Use v0 when serializing an unhandled ApiVersion response
-        if (isUnsupportedApiVersionsRequest())
+        if (isUnsupportedApiVersionsRequest(header))
             return 0;
         return header.apiVersion();
     }

Diff for: core/src/main/scala/kafka/network/RequestChannel.scala

@@ -63,7 +63,8 @@ object RequestChannel extends Logging {
                 val memoryPool: MemoryPool,
                 @volatile var buffer: ByteBuffer,
                 metrics: RequestChannelMetrics,
-                val envelope: Option[RequestChannel.Request] = None) extends BaseRequest {
+                val envelope: Option[RequestChannel.Request] = None,
+                val requestAndSize: Option[RequestAndSize] = None) extends BaseRequest {
     // These need to be volatile because the readers are in the network thread and the writers are in the request
     // handler threads or the purgatory threads
     @volatile var requestDequeueTimeNanos: Long = -1L
@@ -79,7 +80,7 @@ object RequestChannel extends Logging {
 
     val session: Session = new Session(context.principal, context.clientAddress)
 
-    private val bodyAndSize: RequestAndSize = context.parseRequest(buffer)
+    private val bodyAndSize: RequestAndSize = requestAndSize.getOrElse(context.parseRequest(buffer))
 
     // This is constructed on creation of a Request so that the JSON representation is computed before the request is
     // processed by the api layer. Otherwise, a ProduceRequest can occur without its data (ie. it goes into purgatory).

Diff for: core/src/main/scala/kafka/network/SocketServer.scala

@@ -40,7 +40,7 @@ import org.apache.kafka.common.metrics.stats.{Avg, CumulativeSum, Meter, Rate}
 import org.apache.kafka.common.network.KafkaChannel.ChannelMuteEvent
 import org.apache.kafka.common.network.{ChannelBuilder, ChannelBuilders, ClientInformation, KafkaChannel, ListenerName, ListenerReconfigurable, NetworkSend, Selectable, Send, ServerConnectionId, Selector => KSelector}
 import org.apache.kafka.common.protocol.ApiKeys
-import org.apache.kafka.common.requests.{ApiVersionsRequest, RequestContext, RequestHeader}
+import org.apache.kafka.common.requests.{ApiVersionsRequest, RequestAndSize, RequestContext, RequestHeader}
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.{KafkaThread, LogContext, Time, Utils}
 import org.apache.kafka.common.{Endpoint, KafkaException, MetricName, Reconfigurable}
@@ -1133,23 +1133,27 @@ private[kafka] class Processor(
                 expiredConnectionsKilledCount.record(null, 1, 0)
               } else {
                 val connectionId = receive.source
-                val context = new RequestContext(header, connectionId, channel.socketAddress, Optional.of(channel.socketPort()),
-                  channel.principal, listenerName, securityProtocol, channel.channelMetadataRegistry.clientInformation,
-                  isPrivilegedListener, channel.principalSerde)
-
-                val req = new RequestChannel.Request(processor = id, context = context,
-                  startTimeNanos = nowNanos, memoryPool, receive.payload, requestChannel.metrics, None)
-
+                var requestAndSize: Option[RequestAndSize] = None
                 // KIP-511: ApiVersionsRequest is intercepted here to catch the client software name
                 // and version. It is done here to avoid wiring things up to the api layer.
                 if (header.apiKey == ApiKeys.API_VERSIONS) {
-                  val apiVersionsRequest = req.body[ApiVersionsRequest]
+                  val result = RequestContext.staticParseRequest(header, receive.payload, connectionId, listenerName, channel.principal)
+                  val apiVersionsRequest = result.request.asInstanceOf[ApiVersionsRequest]
                   if (apiVersionsRequest.isValid) {
                     channel.channelMetadataRegistry.registerClientInformation(new ClientInformation(
                       apiVersionsRequest.data.clientSoftwareName,
                       apiVersionsRequest.data.clientSoftwareVersion))
                   }
+                  requestAndSize = Some(result)
                 }
+
+                val context = new RequestContext(header, connectionId, channel.socketAddress, Optional.of(channel.socketPort()),
+                  channel.principal, listenerName, securityProtocol, channel.channelMetadataRegistry.clientInformation,
+                  isPrivilegedListener, channel.principalSerde)
+
+                val req = new RequestChannel.Request(processor = id, context = context,
+                  startTimeNanos = nowNanos, memoryPool, receive.payload, requestChannel.metrics, None, requestAndSize)
+
                 requestChannel.sendRequest(req)
                 selector.mute(connectionId)
                 handleChannelMuteEvent(connectionId, ChannelMuteEvent.REQUEST_RECEIVED)

Diff for: core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java

@@ -527,7 +527,7 @@ private RequestChannel.Request buildRequest(AbstractRequest request,
                 listenerName, SecurityProtocol.SSL, ClientInformation.EMPTY, false,
                 Optional.of(kafkaPrincipalSerde));
         return new RequestChannel.Request(1, context, 0, MemoryPool.NONE, buffer,
-                requestChannelMetrics, scala.Option.apply(null));
+                requestChannelMetrics, scala.Option.apply(null), scala.Option.apply(null));
     }
 
     KafkaConfig createKafkaDefaultConfig() {

Diff for: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java

@@ -226,7 +226,7 @@ private RequestChannel.Request buildAllTopicMetadataRequest() {
         RequestContext context = new RequestContext(header, "1", null, principal,
                 ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
                 SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false);
-        return new RequestChannel.Request(1, context, 0, MemoryPool.NONE, bodyBuffer, requestChannelMetrics, Option.empty());
+        return new RequestChannel.Request(1, context, 0, MemoryPool.NONE, bodyBuffer, requestChannelMetrics, Option.empty(), Option.empty());
     }
 
     @Benchmark