KAFKA-17563 Move RequestConvertToJson to server module (apache#17223)

Reviewers: Chia-Ping Tsai <chia7712@apache.org>
ahuang98 · Sep 26, 2024 · 18340c9 · 18340c9
1 parent f8acfa5
commit 18340c9
Show file tree

Hide file tree

Showing 14 changed files with 968 additions and 368 deletions.
diff --git a/checkstyle/import-control-server.xml b/checkstyle/import-control-server.xml
@@ -99,4 +99,8 @@
     <allow pkg="org.apache.kafka.server.authorizer" />
   </subpackage>
 
+  <subpackage name="network">
+    <allow pkg="com.fasterxml.jackson" />
+  </subpackage>
+
 </import-control>
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
@@ -50,6 +50,7 @@
     <!-- server tests -->
     <suppress checks="MethodLength|JavaNCSS|NPath" files="DescribeTopicPartitionsRequestHandlerTest.java"/>
     <suppress checks="CyclomaticComplexity" files="ListConsumerGroupTest.java"/>
+    <suppress checks="ClassFanOutComplexity|CyclomaticComplexity|MethodLength|ParameterNumber|JavaNCSS|ImportControl" files="RequestConvertToJson.java"/>
 
     <!-- Clients -->
     <suppress id="dontUseSystemExit"

diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -36,8 +36,10 @@ import org.apache.kafka.network.Session
 import org.apache.kafka.network.metrics.{RequestChannelMetrics, RequestMetrics}
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
+import org.apache.kafka.network.RequestConvertToJson
 
 import scala.jdk.CollectionConverters._
+import scala.compat.java8.OptionConverters._
 import scala.reflect.ClassTag
 
 object RequestChannel extends Logging {
@@ -249,7 +251,7 @@ object RequestChannel extends Logging {
       recordNetworkThreadTimeCallback.foreach(record => record(networkThreadTimeNanos))
 
       if (isRequestLoggingEnabled) {
-        val desc = RequestConvertToJson.requestDescMetrics(header, requestLog, response.responseLog,
+        val desc = RequestConvertToJson.requestDescMetrics(header, requestLog.asJava, response.responseLog.asJava,
           context, session, isForwarded,
           totalTimeMs, requestQueueTimeMs, apiLocalTimeMs,
           apiRemoteTimeMs, apiThrottleTimeMs, responseQueueTimeMs,

diff --git a/core/src/main/scala/kafka/network/RequestConvertToJson.scala b/core/src/main/scala/kafka/network/RequestConvertToJson.scala
diff --git a/core/src/test/scala/unit/kafka/network/RequestChannelTest.scala b/core/src/test/scala/unit/kafka/network/RequestChannelTest.scala
@@ -35,6 +35,7 @@ import org.apache.kafka.common.requests.AlterConfigsRequest._
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.security.auth.{KafkaPrincipal, KafkaPrincipalSerde, SecurityProtocol}
 import org.apache.kafka.common.utils.{SecurityUtils, Utils}
+import org.apache.kafka.network.RequestConvertToJson
 import org.apache.kafka.network.metrics.RequestChannelMetrics
 import org.apache.kafka.test
 import org.junit.jupiter.api.Assertions._
@@ -49,6 +50,7 @@ import java.nio.ByteBuffer
 import java.util.Collections
 import java.util.concurrent.atomic.AtomicReference
 import scala.collection.{Map, Seq}
+import scala.compat.java8.OptionConverters._
 import scala.jdk.CollectionConverters._
 
 class RequestChannelTest {
@@ -69,7 +71,7 @@ class RequestChannelTest {
       val loggableAlterConfigs = alterConfigs.loggableRequest.asInstanceOf[AlterConfigsRequest]
       val loggedConfig = loggableAlterConfigs.configs.get(resource)
       assertEquals(expectedValues, toMap(loggedConfig))
-      val alterConfigsDesc = RequestConvertToJson.requestDesc(alterConfigs.header, alterConfigs.requestLog, alterConfigs.isForwarded).toString
+      val alterConfigsDesc = RequestConvertToJson.requestDesc(alterConfigs.header, alterConfigs.requestLog.asJava, alterConfigs.isForwarded).toString
       assertFalse(alterConfigsDesc.contains(sensitiveValue), s"Sensitive config logged $alterConfigsDesc")
     }
 
@@ -133,7 +135,7 @@ class RequestChannelTest {
       val loggableAlterConfigs = alterConfigs.loggableRequest.asInstanceOf[IncrementalAlterConfigsRequest]
       val loggedConfig = loggableAlterConfigs.data.resources.find(resource.`type`.id, resource.name).configs
       assertEquals(expectedValues, toMap(loggedConfig))
-      val alterConfigsDesc = RequestConvertToJson.requestDesc(alterConfigs.header, alterConfigs.requestLog, alterConfigs.isForwarded).toString
+      val alterConfigsDesc = RequestConvertToJson.requestDesc(alterConfigs.header, alterConfigs.requestLog.asJava, alterConfigs.isForwarded).toString
       assertFalse(alterConfigsDesc.contains(sensitiveValue), s"Sensitive config logged $alterConfigsDesc")
     }
 

diff --git a/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala b/core/src/test/scala/unit/kafka/network/RequestConvertToJsonTest.scala
@@ -21,91 +21,22 @@ import java.net.InetAddress
 import java.nio.ByteBuffer
 import com.fasterxml.jackson.databind.node.{BooleanNode, DoubleNode, JsonNodeFactory, LongNode, ObjectNode, TextNode}
 import kafka.network
-import kafka.network.RequestConvertToJson.requestHeaderNode
 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, Errors, MessageUtil}
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
+import org.apache.kafka.network.RequestConvertToJson
 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 java.util.Collections
-import scala.collection.mutable.ArrayBuffer
+import scala.compat.java8.OptionConverters._
 
 class RequestConvertToJsonTest {
 
-  @Test
-  def testAllRequestTypesHandled(): Unit = {
-    val unhandledKeys = ArrayBuffer[String]()
-    ApiKeys.values().foreach { key => {
-      val version: Short = key.latestVersion()
-      val message = key match {
-        case ApiKeys.DESCRIBE_ACLS =>
-          ApiMessageType.fromApiKey(key.id).newRequest().asInstanceOf[DescribeAclsRequestData]
-            .setPatternTypeFilter(1).setResourceTypeFilter(1).setPermissionType(1).setOperation(1)
-        case _ =>
-          ApiMessageType.fromApiKey(key.id).newRequest()
-      }
-
-      val bytes = MessageUtil.toByteBuffer(message, version)
-      val req = AbstractRequest.parseRequest(key, version, bytes).request
-      try {
-        RequestConvertToJson.request(req)
-      } catch {
-        case _ : IllegalStateException => unhandledKeys += key.toString
-      }
-    }}
-    assertEquals(ArrayBuffer.empty, unhandledKeys, "Unhandled request keys")
-  }
-
-  @Test
-  def testAllApiVersionsResponseHandled(): Unit = {
-
-    ApiKeys.values().foreach { key => {
-      val unhandledVersions = ArrayBuffer[java.lang.Short]()
-      key.allVersions().forEach { version => {
-        val message = key match {
-          // Specify top-level error handling for verifying compatibility across versions
-          case ApiKeys.DESCRIBE_LOG_DIRS =>
-            ApiMessageType.fromApiKey(key.id).newResponse().asInstanceOf[DescribeLogDirsResponseData]
-              .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code())
-          case _ =>
-            ApiMessageType.fromApiKey(key.id).newResponse()
-        }
-
-        val bytes = MessageUtil.toByteBuffer(message, version)
-        val response = AbstractResponse.parseResponse(key, bytes, version)
-        try {
-          RequestConvertToJson.response(response, version)
-        } catch {
-          case _ : IllegalStateException => unhandledVersions += version
-        }}
-      }
-      assertEquals(ArrayBuffer.empty, unhandledVersions, s"API: ${key.toString} - Unhandled request versions")
-    }}
-  }
-
-  @Test
-  def testAllResponseTypesHandled(): Unit = {
-    val unhandledKeys = ArrayBuffer[String]()
-    ApiKeys.values().foreach { key => {
-      val version: Short = key.latestVersion()
-      val message = ApiMessageType.fromApiKey(key.id).newResponse()
-      val bytes = MessageUtil.toByteBuffer(message, version)
-      val res = AbstractResponse.parseResponse(key, bytes, version)
-      try {
-        RequestConvertToJson.response(res, version)
-      } catch {
-        case _ : IllegalStateException => unhandledKeys += key.toString
-      }
-    }}
-    assertEquals(ArrayBuffer.empty, unhandledKeys, "Unhandled response keys")
-  }
-
   @Test
   def testRequestHeaderNode(): Unit = {
     val alterIsrRequest = new AlterPartitionRequest(new AlterPartitionRequestData(), 0)
@@ -135,30 +66,17 @@ class RequestConvertToJsonTest {
     assertEquals(expectedNode, actualNode)
   }
 
-  @Test
-  def testClientInfoNode(): Unit = {
-    val clientInfo = new ClientInformation("name", "1")
-
-    val expectedNode = new ObjectNode(JsonNodeFactory.instance)
-    expectedNode.set("softwareName", new TextNode(clientInfo.softwareName))
-    expectedNode.set("softwareVersion", new TextNode(clientInfo.softwareVersion))
-
-    val actualNode = RequestConvertToJson.clientInfoNode(clientInfo)
-
-    assertEquals(expectedNode, actualNode)
-  }
-
   @Test
   def testRequestDesc(): Unit = {
     val alterIsrRequest = new AlterPartitionRequest(new AlterPartitionRequestData(), 0)
     val req = request(alterIsrRequest)
 
     val expectedNode = new ObjectNode(JsonNodeFactory.instance)
     expectedNode.set("isForwarded", if (req.isForwarded) BooleanNode.TRUE else BooleanNode.FALSE)
-    expectedNode.set("requestHeader", requestHeaderNode(req.header))
+    expectedNode.set("requestHeader", RequestConvertToJson.requestHeaderNode(req.header))
     expectedNode.set("request", req.requestLog.getOrElse(new TextNode("")))
 
-    val actualNode = RequestConvertToJson.requestDesc(req.header, req.requestLog, req.isForwarded)
+    val actualNode = RequestConvertToJson.requestDesc(req.header, req.requestLog.asJava, req.isForwarded)
 
     assertEquals(expectedNode, actualNode)
   }
@@ -181,7 +99,7 @@ class RequestConvertToJsonTest {
     val temporaryMemoryBytes = 8
     val messageConversionsTimeMs = 9
 
-    val expectedNode = RequestConvertToJson.requestDesc(req.header, req.requestLog, req.isForwarded).asInstanceOf[ObjectNode]
+    val expectedNode = RequestConvertToJson.requestDesc(req.header, req.requestLog.asJava, req.isForwarded).asInstanceOf[ObjectNode]
     expectedNode.set("response", res.responseLog.getOrElse(new TextNode("")))
     expectedNode.set("connection", new TextNode(req.context.connectionId))
     expectedNode.set("totalTimeMs", new DoubleNode(totalTimeMs))
@@ -198,7 +116,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, req.context, req.session, req.isForwarded,
+    val actualNode = RequestConvertToJson.requestDescMetrics(req.header, req.requestLog.asJava, res.responseLog.asJava, 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
@@ -35,6 +35,7 @@ import org.apache.kafka.common.requests._
 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.network.RequestConvertToJson
 import org.apache.kafka.network.SocketServerConfigs
 import org.apache.kafka.network.metrics.RequestMetrics
 import org.apache.kafka.security.CredentialProvider

diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/FetchRequestBenchmark.java
@@ -17,8 +17,6 @@
 
 package org.apache.kafka.jmh.common;
 
-import kafka.network.RequestConvertToJson;
-
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.network.Send;
@@ -27,6 +25,7 @@
 import org.apache.kafka.common.requests.ByteBufferChannel;
 import org.apache.kafka.common.requests.FetchRequest;
 import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.network.RequestConvertToJson;
 
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;

diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/ListOffsetRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/ListOffsetRequestBenchmark.java
@@ -17,13 +17,12 @@
 
 package org.apache.kafka.jmh.common;
 
-import kafka.network.RequestConvertToJson;
-
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.message.ListOffsetsRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.network.RequestConvertToJson;
 
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;

diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/ProduceRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/common/ProduceRequestBenchmark.java
@@ -17,11 +17,10 @@
 
 package org.apache.kafka.jmh.common;
 
-import kafka.network.RequestConvertToJson;
-
 import org.apache.kafka.common.message.ProduceRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.requests.ProduceRequest;
+import org.apache.kafka.network.RequestConvertToJson;
 
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;

diff --git a/...benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java b/...benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java
@@ -19,7 +19,6 @@
 
 import kafka.coordinator.transaction.TransactionCoordinator;
 import kafka.network.RequestChannel;
-import kafka.network.RequestConvertToJson;
 import kafka.server.AutoTopicCreationManager;
 import kafka.server.ClientQuotaManager;
 import kafka.server.ClientRequestQuotaManager;
@@ -58,6 +57,7 @@
 import org.apache.kafka.image.MetadataDelta;
 import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.image.MetadataProvenance;
+import org.apache.kafka.network.RequestConvertToJson;
 import org.apache.kafka.network.metrics.RequestChannelMetrics;
 import org.apache.kafka.raft.QuorumConfig;
 import org.apache.kafka.server.common.FinalizedFeatures;
@@ -237,7 +237,9 @@ public void testMetadataRequestForAllTopics() {
 
     @Benchmark
     public String testRequestToJson() {
-        return RequestConvertToJson.requestDesc(allTopicMetadataRequest.header(), allTopicMetadataRequest.requestLog(), allTopicMetadataRequest.isForwarded()).toString();
+        Option<com.fasterxml.jackson.databind.JsonNode> option = allTopicMetadataRequest.requestLog();
+        Optional<com.fasterxml.jackson.databind.JsonNode> optional = option.isDefined() ? Optional.of(option.get()) : Optional.empty();
+        return RequestConvertToJson.requestDesc(allTopicMetadataRequest.header(), optional, allTopicMetadataRequest.isForwarded()).toString();
     }
 
     @Benchmark

diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java
@@ -20,7 +20,6 @@
 import kafka.controller.KafkaController;
 import kafka.coordinator.transaction.TransactionCoordinator;
 import kafka.network.RequestChannel;
-import kafka.network.RequestConvertToJson;
 import kafka.server.AutoTopicCreationManager;
 import kafka.server.BrokerFeatures;
 import kafka.server.ClientQuotaManager;
@@ -59,6 +58,7 @@
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.coordinator.group.GroupCoordinator;
+import org.apache.kafka.network.RequestConvertToJson;
 import org.apache.kafka.network.metrics.RequestChannelMetrics;
 import org.apache.kafka.server.common.FinalizedFeatures;
 import org.apache.kafka.server.common.MetadataVersion;
@@ -237,7 +237,9 @@ public void testMetadataRequestForAllTopics() {
 
     @Benchmark
     public String testRequestToJson() {
-        return RequestConvertToJson.requestDesc(allTopicMetadataRequest.header(), allTopicMetadataRequest.requestLog(), allTopicMetadataRequest.isForwarded()).toString();
+        Option<com.fasterxml.jackson.databind.JsonNode> option = allTopicMetadataRequest.requestLog();
+        Optional<com.fasterxml.jackson.databind.JsonNode> optional = option.isDefined() ? Optional.of(option.get()) : Optional.empty();
+        return RequestConvertToJson.requestDesc(allTopicMetadataRequest.header(), optional, allTopicMetadataRequest.isForwarded()).toString();
     }
 
     @Benchmark