simpify logic

Author: TaiJuWu

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

@@ -168,16 +168,6 @@ class KafkaApis(val requestChannel: RequestChannel,
     maybeForwardToController(request, errorHandler)
   }
 
-  private def forwardToControllerOrThrow(request: RequestChannel.Request,
-                                          createException: RequestChannel.Request => Exception
-                                       ): Unit = {
-    def errorHandler(request: RequestChannel.Request): Unit = {
-      throw createException(request)
-    }
-
-    maybeForwardToController(request, errorHandler)
-  }
-
   /**
    * Top-level method that handles all requests and multiplexes to the right api
    */
@@ -218,8 +208,8 @@ class KafkaApis(val requestChannel: RequestChannel,
         case ApiKeys.LIST_GROUPS => handleListGroupsRequest(request).exceptionally(handleError)
         case ApiKeys.SASL_HANDSHAKE => handleSaslHandshakeRequest(request)
         case ApiKeys.API_VERSIONS => handleApiVersionsRequest(request)
-        case ApiKeys.CREATE_TOPICS => forwardToControllerOrThrow(request, KafkaApis.shouldAlwaysForward)
-        case ApiKeys.DELETE_TOPICS => forwardToControllerOrThrow(request, KafkaApis.shouldAlwaysForward)
+        case ApiKeys.CREATE_TOPICS => maybeForwardToController(request, handleCreateTopicsRequest)
+        case ApiKeys.DELETE_TOPICS => maybeForwardToController(request, handleDeleteTopicsRequest)
         case ApiKeys.DELETE_RECORDS => handleDeleteRecordsRequest(request)
         case ApiKeys.INIT_PRODUCER_ID => handleInitProducerIdRequest(request, requestLocal)
         case ApiKeys.OFFSET_FOR_LEADER_EPOCH => handleOffsetForLeaderEpochRequest(request)
@@ -236,7 +226,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         case ApiKeys.ALTER_REPLICA_LOG_DIRS => handleAlterReplicaLogDirsRequest(request)
         case ApiKeys.DESCRIBE_LOG_DIRS => handleDescribeLogDirsRequest(request)
         case ApiKeys.SASL_AUTHENTICATE => handleSaslAuthenticateRequest(request)
-        case ApiKeys.CREATE_PARTITIONS => forwardToControllerOrThrow(request, KafkaApis.shouldAlwaysForward)
+        case ApiKeys.CREATE_PARTITIONS => maybeForwardToController(request, handleCreatePartitionsRequest)
         // Create, renew and expire DelegationTokens must first validate that the connection
         // itself is not authenticated with a delegation token before maybeForwardToController.
         case ApiKeys.CREATE_DELEGATION_TOKEN => handleCreateTokenRequest(request)
@@ -1778,6 +1768,18 @@ class KafkaApis(val requestChannel: RequestChannel,
     requestHelper.sendResponseMaybeThrottle(request, createResponseCallback)
   }
 
+  def handleCreateTopicsRequest(request: RequestChannel.Request): Unit = {
+    throw KafkaApis.shouldAlwaysForward(request)
+  }
+
+  def handleCreatePartitionsRequest(request: RequestChannel.Request): Unit = {
+    throw KafkaApis.shouldAlwaysForward(request)
+  }
+
+  def handleDeleteTopicsRequest(request: RequestChannel.Request): Unit = {
+    throw KafkaApis.shouldAlwaysForward(request)
+  }
+
   def handleDeleteRecordsRequest(request: RequestChannel.Request): Unit = {
     val deleteRecordsRequest = request.body[DeleteRecordsRequest]
 

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -10514,6 +10514,27 @@ class KafkaApisTest extends Logging {
     verifyShouldNeverHandleErrorMessage(kafkaApis.handleEnvelope(_, RequestLocal.withThreadConfinedCaching))
   }
 
+  @Test
+  def testRaftShouldAlwaysForwardCreateTopicsRequest(): Unit = {
+    metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
+    kafkaApis = createKafkaApis(raftSupport = true)
+    verifyShouldAlwaysForwardErrorMessage(kafkaApis.handleCreateTopicsRequest)
+  }
+
+  @Test
+  def testRaftShouldAlwaysForwardCreatePartitionsRequest(): Unit = {
+    metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
+    kafkaApis = createKafkaApis(raftSupport = true)
+    verifyShouldAlwaysForwardErrorMessage(kafkaApis.handleCreatePartitionsRequest)
+  }
+
+  @Test
+  def testRaftShouldAlwaysForwardDeleteTopicsRequest(): Unit = {
+    metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)
+    kafkaApis = createKafkaApis(raftSupport = true)
+    verifyShouldAlwaysForwardErrorMessage(kafkaApis.handleDeleteTopicsRequest)
+  }
+
   @Test
   def testRaftShouldAlwaysForwardCreateAcls(): Unit = {
     metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () => KRaftVersion.KRAFT_VERSION_0)