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

@@ -41,7 +41,9 @@ public class RequestContext implements AuthorizableRequestContext {
     public final KafkaPrincipal principal;
     public final ListenerName listenerName;
     public final SecurityProtocol securityProtocol;
-    public final ClientInformation clientInformation;
+    // The client information can be updated if the request is ApiVersionRequest,
+    // so the client information will not be unknown for ApiVersionRequest.
+    public ClientInformation clientInformation;
     public final boolean fromPrivilegedListener;
     public final Optional<KafkaPrincipalSerde> principalSerde;
 

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

@@ -1145,6 +1145,7 @@ private[kafka] class Processor(
                     channel.channelMetadataRegistry.registerClientInformation(new ClientInformation(
                       apiVersionsRequest.data.clientSoftwareName,
                       apiVersionsRequest.data.clientSoftwareVersion))
+                    context.clientInformation = channel.channelMetadataRegistry.clientInformation
                   }
                 }
                 requestChannel.sendRequest(req)

Diff for: core/src/test/scala/unit/kafka/network/SocketServerTest.scala

@@ -274,8 +274,8 @@ class SocketServerTest {
     sendRequest(plainSocket, apiVersionRequestBytes(clientId, version))
     var receivedReq = receiveRequest(server.dataPlaneRequestChannel)
 
-    assertEquals(ClientInformation.UNKNOWN_NAME_OR_VERSION, receivedReq.context.clientInformation.softwareName)
-    assertEquals(ClientInformation.UNKNOWN_NAME_OR_VERSION, receivedReq.context.clientInformation.softwareVersion)
+    assertEquals(expectedClientSoftwareName, receivedReq.context.clientInformation.softwareName)
+    assertEquals(expectedClientSoftwareVersion, receivedReq.context.clientInformation.softwareVersion)
 
     server.dataPlaneRequestChannel.sendNoOpResponse(receivedReq)