adding client id to observer

linkedin · Mar 12, 2024 · ee9c4e2 · ee9c4e2
1 parent 348c434
commit ee9c4e2
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -1952,7 +1952,7 @@ class KafkaApis(val requestChannel: RequestChannel,
           hasNonXinfraClient = true;
         }
       }
-      observer.checkClientLibrary(hasNonXinfraClient)
+      observer.checkClientLibrary(hasNonXinfraClient, request.context.clientId())
       if (!expectedClientSoftwareNames.contains(softwareName)) {
         val clientIdentity = request.context.clientId() + " " + request.context.clientAddress() + " " + request.context.principal()
         unofficialClientsCache.get(clientIdentity)

diff --git a/core/src/main/scala/kafka/server/Observer.scala b/core/src/main/scala/kafka/server/Observer.scala
@@ -61,9 +61,10 @@ trait Observer extends Configurable {
   /**
     * Check client library. Users pass in the client library and version so the client type can be checked
     *
-    * @param hasNonXinfraClient  flag to determine if the topic has been accessed by a non xinfra client
+   * @param hasNonXinfraClient  flag to determine if the topic has been accessed by a non xinfra client
+   * @param clientId The clientId for this specific request
     */
-  def checkClientLibrary(hasNonXinfraClient: Boolean): Unit
+  def checkClientLibrary(hasNonXinfraClient: Boolean, clientId: String): Unit
 
   /**
     * Close the observer with timeout.