KAFKA-19931 Align broker and controller behavior for the Admin.incrementalAlterConfigs API

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

@@ -58,7 +58,7 @@ import scala.jdk.CollectionConverters._
  * KIP-412 added support for changing log4j log levels via IncrementalAlterConfigs, but
  * not via the original AlterConfigs. In retrospect, this would have been better off as a
  * separate RPC, since the semantics are quite different. In particular, KIP-226 configs
- * are stored durably and persist across broker restarts, but KIP-412 log4j levels do not. 
+ * are stored durably and persist across broker restarts, but KIP-412 log4j levels do not.
  * However, we have to handle it here now in order to maintain compatibility.
  *
  * Configuration processing is split into two parts.
@@ -112,48 +112,33 @@ class ConfigAdminManager(nodeId: Int,
     })
     request.resources().forEach(resource => {
       if (!results.containsKey(resource)) {
-        val resourceType = ConfigResource.Type.forId(resource.resourceType())
-        val configResource = new ConfigResource(resourceType, resource.resourceName())
-        try {
-          if (containsDuplicates(resource.configs().asScala.map(_.name()))) {
-            throw new InvalidRequestException("Error due to duplicate config keys")
-          }
-          val nullUpdates = new util.ArrayList[String]()
-          resource.configs().forEach { config =>
-            if (config.configOperation() != AlterConfigOp.OpType.DELETE.id() &&
-              config.value() == null) {
-              nullUpdates.add(config.name())
+        processConfigResource(
+          resource,
+          () => {
+            runtimeLoggerManager.applyChangesForResource(
+              authorize(ResourceType.CLUSTER, Resource.CLUSTER_NAME),
+              request.validateOnly(),
+              resource)
+            results.put(resource, ApiError.NONE)
+          },
+          configResource => {
+            // The resource name must be either blank (if setting a cluster config) or
+            // the ID of this specific broker.
+            if (configResource.name().nonEmpty) {
+              validateResourceNameIsCurrentNodeId(resource.resourceName())
             }
-          }
-          if (!nullUpdates.isEmpty) {
-            throw new InvalidRequestException("Null value not supported for : " +
-              String.join(", ", nullUpdates))
-          }
-          resourceType match {
-            case BROKER_LOGGER =>
-              runtimeLoggerManager.applyChangesForResource(
-                  authorize(ResourceType.CLUSTER, Resource.CLUSTER_NAME),
-                  request.validateOnly(),
-                  resource)
-              results.put(resource, ApiError.NONE)
-            case BROKER =>
-              // The resource name must be either blank (if setting a cluster config) or
-              // the ID of this specific broker.
-              if (configResource.name().nonEmpty) {
-                validateResourceNameIsCurrentNodeId(resource.resourceName())
-              }
-              validateBrokerConfigChange(resource, configResource)
-            case TOPIC | CLIENT_METRICS | GROUP =>
-            // Nothing to do.
-            case _ =>
-              throw new InvalidRequestException(s"Unknown resource type ${resource.resourceType().toInt}")
-          }
-        } catch {
-          case t: Throwable =>
+            validateBrokerConfigChange(resource, configResource)
+          },
+          _ => {
+            // Nothing to do for TOPIC, CLIENT_METRICS or GROUP type in Broker,
+            // and UNKNOWN type is already handled in validateUnknownConfigTypeError.
+          },
+          (configResource, t) => {
             val err = ApiError.fromThrowable(t)
             error(s"Error preprocessing incrementalAlterConfigs request on $configResource", t)
             results.put(resource, err)
-        }
+          }
+        )
       }
     })
     results
@@ -451,4 +436,57 @@ object ConfigAdminManager {
       }
     }
   }
+
+  def processConfigResource(
+    resource: IncrementalAlterConfigsRequestData.AlterConfigsResource,
+    onBrokerLogger: () => Unit,
+    onBroker: ConfigResource => Unit,
+    onOtherTypes: ConfigResource => Unit,
+    onError: (ConfigResource, Throwable) => Unit
+  ): Unit = {
+    val resourceType = ConfigResource.Type.forId(resource.resourceType())
+    val configResource = new ConfigResource(resourceType, resource.resourceName())
+    try {
+      if (containsDuplicates(resource.configs().asScala.map(_.name()))) {
+        throw new InvalidRequestException("Error due to duplicate config keys")
+      }
+      validateNullValue(resource)
+      validateUnknownConfigTypeError(configResource, resource)
+      resourceType match {
+        case BROKER_LOGGER => onBrokerLogger()
+        case BROKER => onBroker(configResource)
+        case _ => onOtherTypes(configResource)
+      }
+    } catch {
+      case t: Throwable => onError(configResource, t)
+    }
+  }
+
+  private def validateNullValue(
+    resource: IncrementalAlterConfigsRequestData.AlterConfigsResource,
+  ): Unit = {
+    val nullUpdates = new util.ArrayList[String]()
+    resource.configs.forEach { config =>
+      if (config.configOperation() != AlterConfigOp.OpType.DELETE.id() &&
+        config.value() == null) {
+        nullUpdates.add(config.name())
+      }
+    }
+    if (!nullUpdates.isEmpty) {
+      val exception = new InvalidRequestException("Null value not supported for : " + String.join(", ", nullUpdates))
+      log.error(s"Error on processing incrementalAlterConfigs request on ${resource.resourceName()}", exception)
+      throw exception
+    }
+  }
+
+  private def validateUnknownConfigTypeError(
+    configResource: ConfigResource,
+    resource: IncrementalAlterConfigsRequestData.AlterConfigsResource,
+  ): Unit = {
+    if (configResource.`type`().equals(ConfigResource.Type.UNKNOWN)) {
+      val exception = new InvalidRequestException(s"Unknown resource type ${resource.resourceType()}.")
+      log.error(s"Error on processing request on ${configResource}", exception)
+      throw exception
+    }
+  }
 }

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

@@ -721,45 +721,35 @@ class ControllerApis(
       util.Map[String, Entry[AlterConfigOp.OpType, String]]]()
     val brokerLoggerResponses = new util.ArrayList[AlterConfigsResourceResponse](1)
     alterConfigsRequest.data.resources.forEach { resource =>
-      val configResource = new ConfigResource(
-        ConfigResource.Type.forId(resource.resourceType), resource.resourceName())
-      if (configResource.`type`().equals(ConfigResource.Type.BROKER_LOGGER)) {
-        val apiError = try {
+      ConfigAdminManager.processConfigResource(
+        resource,
+        () => {
           runtimeLoggerManager.applyChangesForResource(
             authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME),
             alterConfigsRequest.data().validateOnly(),
             resource)
-          ApiError.NONE
-        } catch {
-          case t: Throwable => ApiError.fromThrowable(t)
+          brokerLoggerResponses.add(new AlterConfigsResourceResponse()
+            .setResourceName(resource.resourceName())
+            .setResourceType(resource.resourceType())
+            .setErrorCode(ApiError.NONE.error().code())
+            .setErrorMessage(null))
+        },
+        configResource => {
+          addConfigChangesOrHandleDuplicate(configResource, resource, duplicateResources, configChanges, response)
+        },
+        configResource => {
+          addConfigChangesOrHandleDuplicate(configResource, resource, duplicateResources, configChanges, response)
+        },
+        (_, t) => {
+          val err = ApiError.fromThrowable(t)
+          error(s"Error on processing incrementalAlterConfigs request on ${resource.resourceName()}", t)
+          response.responses().add(new AlterConfigsResourceResponse()
+            .setErrorCode(err.error().code())
+            .setErrorMessage(err.messageWithFallback())
+            .setResourceName(resource.resourceName())
+            .setResourceType(resource.resourceType()))
         }
-        brokerLoggerResponses.add(new AlterConfigsResourceResponse().
-          setResourceName(resource.resourceName()).
-          setResourceType(resource.resourceType()).
-          setErrorCode(apiError.error().code()).
-          setErrorMessage(if (apiError.isFailure) apiError.messageWithFallback() else null))
-      } else if (configResource.`type`().equals(ConfigResource.Type.UNKNOWN)) {
-        response.responses().add(new AlterConfigsResourceResponse().
-          setErrorCode(UNSUPPORTED_VERSION.code()).
-          setErrorMessage("Unknown resource type " + resource.resourceType() + ".").
-          setResourceName(resource.resourceName()).
-          setResourceType(resource.resourceType()))
-      } else if (!duplicateResources.contains(configResource)) {
-        val altersByName = new util.HashMap[String, Entry[AlterConfigOp.OpType, String]]()
-        resource.configs.forEach { config =>
-          altersByName.put(config.name, new util.AbstractMap.SimpleEntry[AlterConfigOp.OpType, String](
-            AlterConfigOp.OpType.forId(config.configOperation), config.value))
-        }
-        if (configChanges.put(configResource, altersByName) != null) {
-          duplicateResources.add(configResource)
-          configChanges.remove(configResource)
-          response.responses().add(new AlterConfigsResourceResponse().
-            setErrorCode(INVALID_REQUEST.code()).
-            setErrorMessage("Duplicate resource.").
-            setResourceName(resource.resourceName()).
-            setResourceType(resource.resourceType()))
-        }
-      }
+      )
     }
     val iterator = configChanges.keySet().iterator()
     while (iterator.hasNext) {
@@ -792,6 +782,32 @@ class ControllerApis(
       }
   }
 
+  private def addConfigChangesOrHandleDuplicate(
+    configResource: ConfigResource,
+    resource: IncrementalAlterConfigsRequestData.AlterConfigsResource,
+    duplicateResources: util.HashSet[ConfigResource],
+    configChanges: util.HashMap[ConfigResource, util.Map[String, Entry[AlterConfigOp.OpType, String]]],
+    response: IncrementalAlterConfigsResponseData
+  ): Unit = {
+    if (!duplicateResources.contains(configResource)) {
+      val altersByName = new util.HashMap[String, Entry[AlterConfigOp.OpType, String]]()
+      resource.configs.forEach { config =>
+        altersByName.put(config.name, new util.AbstractMap.SimpleEntry[AlterConfigOp.OpType, String](
+          AlterConfigOp.OpType.forId(config.configOperation), config.value))
+      }
+
+      if (configChanges.put(configResource, altersByName) != null) {
+        duplicateResources.add(configResource)
+        configChanges.remove(configResource)
+        response.responses().add(new AlterConfigsResourceResponse()
+          .setErrorCode(INVALID_REQUEST.code())
+          .setErrorMessage("Duplicate resource.")
+          .setResourceName(resource.resourceName())
+          .setResourceType(resource.resourceType()))
+      }
+    }
+  }
+
   private def handleCreatePartitions(request: RequestChannel.Request): CompletableFuture[Unit] = {
     def filterAlterAuthorizedTopics(topics: Iterable[String]): Set[String] = {
       authHelper.filterByAuthorized(request.context, ALTER, TOPIC, topics)(n => n)

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

@@ -19,7 +19,6 @@ package kafka.server
 
 import java.util
 import java.util.Collections
-
 import kafka.utils.TestUtils
 import org.apache.kafka.clients.admin.AlterConfigOp.OpType
 import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, TOPIC, UNKNOWN}
@@ -39,7 +38,8 @@ import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.protocol.Errors.{INVALID_REQUEST, NONE}
 import org.apache.kafka.common.requests.ApiError
 import org.apache.kafka.metadata.MockConfigRepository
-import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertFalse, assertThrows, assertTrue}
+import org.junit.jupiter.api.function.Executable
 import org.junit.jupiter.api.{Assertions, Test}
 import org.slf4j.LoggerFactory
 
@@ -393,7 +393,7 @@ class ConfigAdminManagerTest {
     val manager = newConfigAdminManager(1)
     val unknown = unknownIncremental()
     assertEquals(Collections.singletonMap(unknown,
-      new ApiError(INVALID_REQUEST, "Unknown resource type 0")),
+      new ApiError(INVALID_REQUEST, "Unknown resource type 0.")),
         manager.preprocess(new IncrementalAlterConfigsRequestData().
         setResources(new IAlterConfigsResourceCollection(util.Arrays.asList(
           unknown).iterator())),
@@ -419,4 +419,101 @@ class ConfigAdminManagerTest {
     assertFalse(ConfigAdminManager.containsDuplicates(Seq("foo", "bar", "baz")))
     assertTrue(ConfigAdminManager.containsDuplicates(Seq("foo", "bar", "baz", "foo")))
   }
+
+  @Test
+  def testProcessConfigResourceWithNulls(): Unit = {
+    val brokerLogger2 = brokerLogger2Incremental()
+    new IncrementalAlterConfigsRequestData()
+      .setResources(new IAlterConfigsResourceCollection(
+        util.List.of(brokerLogger2).iterator())
+      ).resources().forEach(resource => {
+        val invalidRequestException = assertThrows(classOf[InvalidRequestException],
+          () => ConfigAdminManager.processConfigResource(
+            resource,
+            () => {},
+            _ => {},
+            _ => {},
+            (_, t) => {
+              throw t
+            })
+        )
+        assertEquals(s"Null value not supported for : ${logger.getName}", invalidRequestException.getMessage)
+      })
+  }
+
+  @Test
+  def testProcessConfigResourceWithUnknownConfigType(): Unit = {
+    val config = new IAlterableConfig().setName(logger.getName)
+      .setValue("INFO")
+      .setConfigOperation(OpType.SET.id())
+    new IncrementalAlterConfigsRequestData()
+      .setResources(new IAlterConfigsResourceCollection(
+        util.List.of(new IAlterConfigsResource()
+          .setResourceName("1")
+          .setResourceType(UNKNOWN.id())
+          .setConfigs(new IAlterableConfigCollection(
+            util.List.of(config).iterator()))).iterator()))
+      .resources().forEach(resource => {
+        val invalidRequestException = assertThrows(classOf[InvalidRequestException],
+          () => ConfigAdminManager.processConfigResource(
+            resource,
+            () => {},
+            _ => {},
+            _ => {},
+            (_, t) => {
+              throw t
+            })
+        )
+        assertEquals(s"Unknown resource type ${UNKNOWN.id()}.", invalidRequestException.getMessage)
+      })
+  }
+
+  @Test
+  def testProcessConfigResourceWithDuplicates(): Unit = {
+    val config = new IAlterableConfig().setName(logger.getName)
+      .setValue("INFO")
+      .setConfigOperation(OpType.SET.id())
+    val config1 = new IAlterableConfig().setName(logger.getName)
+      .setValue("ERROR")
+      .setConfigOperation(OpType.SET.id())
+    new IncrementalAlterConfigsRequestData()
+      .setResources(new IAlterConfigsResourceCollection(
+        util.List.of(new IAlterConfigsResource()
+          .setResourceName("1")
+          .setResourceType(BROKER_LOGGER.id)
+          .setConfigs(new IAlterableConfigCollection(
+            util.List.of(config, config1).iterator()))).iterator()))
+      .resources().forEach(resource => {
+        val invalidRequestException = assertThrows(classOf[InvalidRequestException],
+          () => ConfigAdminManager.processConfigResource(
+            resource,
+            () => {},
+            _ => {},
+            _ => {},
+            (_, t) => {
+              throw t
+            })
+        )
+        assertEquals("Error due to duplicate config keys", invalidRequestException.getMessage)
+      })
+  }
+
+  @Test
+  def testProcessConfigResource(): Unit = {
+    val brokerLogger1 = brokerLogger1Incremental()
+    new IncrementalAlterConfigsRequestData()
+      .setResources(new IAlterConfigsResourceCollection(
+        util.List.of(brokerLogger1).iterator())
+      ).resources().forEach(resource => {
+        val processConfigResource: Executable = () => ConfigAdminManager.processConfigResource(
+          resource,
+          () => {},
+          _ => {},
+          _ => {},
+          (_, t) => {
+            throw t
+          })
+        assertDoesNotThrow(processConfigResource)
+      })
+  }
 }