adding observer code in kafka apis

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -1945,7 +1945,15 @@ class KafkaApis(val requestChannel: RequestChannel,
     if (config.unofficialClientLoggingEnable) {
       // Check if the last part of clientSoftwareName (after commitId) is an unexpected software name
       val softwareName = apiVersionRequest.data.clientSoftwareName().split("-").last
-      if (!config.expectedClientSoftwareNames.contains(softwareName)) {
+      val expectedClientSoftwareNames = config.expectedClientSoftwareNames
+      var hasNonXinfraClient = false
+      expectedClientSoftwareNames.forEach { clientName =>
+        if (!clientName.toLowerCase.contains("xinfra")) {
+          hasNonXinfraClient = true;
+        }
+      }
+      observer.checkClientLibrary(hasNonXinfraClient)
+      if (!expectedClientSoftwareNames.contains(softwareName)) {
         val clientIdentity = request.context.clientId() + " " + request.context.clientAddress() + " " + request.context.principal()
         unofficialClientsCache.get(clientIdentity)
         warn(s"received ApiVersionsRequest from user with unofficial client type. clientId clientAddress principal = $clientIdentity")

core/src/main/scala/kafka/server/Observer.scala

@@ -19,6 +19,8 @@ package kafka.server
 
 import java.util.concurrent.TimeUnit
 import kafka.network.RequestChannel
+import java.util
+
 import kafka.utils.{CoreUtils, Logging}
 import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ProduceRequest, RequestContext}
 import org.apache.kafka.common.Configurable
@@ -61,9 +63,9 @@ trait Observer extends Configurable {
   /**
     * Check client library. Users pass in the client library and version so the client type can be checked
     *
-    * @param clientLibrary  client library name and version
+    * @param hasNonXinfraClient  flag to determine if the topic has been accessed by a non xinfra client
     */
-  def checkClientLibrary(clientLibrary: String): Unit
+  def checkClientLibrary(hasNonXinfraClient: Boolean): Unit
 
   /**
     * Close the observer with timeout.