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 | 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)
+    }
+  }
+
+  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
+    }
+  }
+
+  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,40 @@ 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 {
-          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.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()))
+      ConfigAdminManager.processConfigResource(
+        resource,
+        onBrokerLogger = () => {
+          val apiError = try {
+            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.error().code())
+            .setErrorMessage(if (apiError.isFailure) apiError.messageWithFallback() else null))
+        },
+        configResource => {
+          addConfigChangesOrHandleDuplicate(configResource, resource, duplicateResources, configChanges, response)
+        },
+        configResource => {
+          addConfigChangesOrHandleDuplicate(configResource, resource, duplicateResources, configChanges, response)
+        },
+        onError = (_, 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()))
         }
-      }
+      )
     }
     val iterator = configChanges.keySet().iterator()
     while (iterator.hasNext) {
@@ -792,6 +787,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/ControllerApisTest.scala

@@ -652,7 +652,7 @@ class ControllerApisTest {
         setResourceName("3").
         setResourceType(ConfigResource.Type.BROKER.id()),
       new AlterConfigsResourceResponse().
-        setErrorCode(UNSUPPORTED_VERSION.code()).
+        setErrorCode(INVALID_REQUEST.code()).
         setErrorMessage("Unknown resource type 124.").
         setResourceName("foo").
         setResourceType(124.toByte),
@@ -679,6 +679,38 @@ class ControllerApisTest {
       response.data().responses().asScala.toSet)
   }
 
+  @Test
+  def testInvalidIncrementalAlterConfigsWithNullResources(): Unit = {
+    val requestData = new IncrementalAlterConfigsRequestData().setResources(
+      new AlterConfigsResourceCollection(util.Arrays.asList(
+        new AlterConfigsResource().
+          setResourceName("3").
+          setResourceType(ConfigResource.Type.BROKER.id()).
+          setConfigs(new AlterableConfigCollection(util.Arrays.asList(new AlterableConfig().
+            setName("my.custom.config").
+            setValue(null).
+            setConfigOperation(AlterConfigOp.OpType.SET.id())).iterator()))
+      ).iterator()))
+    val request = buildRequest(new IncrementalAlterConfigsRequest.Builder(requestData).build(0))
+    controllerApis = createControllerApis(None, new MockController.Builder().build())
+    controllerApis.handleIncrementalAlterConfigs(request)
+    val capturedResponse: ArgumentCaptor[AbstractResponse] =
+      ArgumentCaptor.forClass(classOf[AbstractResponse])
+    verify(requestChannel).sendResponse(
+      ArgumentMatchers.eq(request),
+      capturedResponse.capture(),
+      ArgumentMatchers.eq(None))
+    assertNotNull(capturedResponse.getValue)
+    val response = capturedResponse.getValue.asInstanceOf[IncrementalAlterConfigsResponse]
+    assertEquals(Set(
+      new AlterConfigsResourceResponse().
+        setErrorCode(INVALID_REQUEST.code()).
+        setErrorMessage("Null value not supported for : my.custom.config").
+        setResourceName("3").
+        setResourceType(ConfigResource.Type.BROKER.id())),
+      response.data().responses().asScala.toSet)
+  }
+
   @Test
   def testUnauthorizedHandleAlterPartitionReassignments(): Unit = {
     assertThrows(classOf[ClusterAuthorizationException], () => {

server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java

@@ -357,4 +357,39 @@ public void testDescribeConfigs(ClusterInstance clusterInstance) throws Exceptio
             assertEquals("2", configEntry.value());
         }
     }
+
+    @ClusterTest
+    public void testIncrementalAlterNullValueConfigsByControllers(ClusterInstance clusterInstance) throws Exception {
+        testIncrementalAlterNullValueConfigs(clusterInstance, true);
+    }
+
+    @ClusterTest
+    public void testIncrementalAlterNullValueConfigs(ClusterInstance clusterInstance) throws Exception {
+        testIncrementalAlterNullValueConfigs(clusterInstance, false);
+    }
+
+    private void testIncrementalAlterNullValueConfigs(ClusterInstance clusterInstance, boolean usingBootstrapControllers) throws Exception {
+        try (Admin admin = Admin.create(adminConfig(clusterInstance, usingBootstrapControllers))) {
+            int nodeId = usingBootstrapControllers ?
+                    clusterInstance.controllers().values().iterator().next().config().nodeId() :
+                    clusterInstance.brokers().values().iterator().next().config().nodeId();
+            ConfigResource nodeResource = new ConfigResource(BROKER, "" + nodeId);
+            Map<ConfigResource, Collection<AlterConfigOp>> alterations = Map.of(
+                nodeResource,
+                List.of(
+                    new AlterConfigOp(new ConfigEntry("my.custom.config", null), AlterConfigOp.OpType.SET),
+                    new AlterConfigOp(new ConfigEntry("my.custom.config1", null), AlterConfigOp.OpType.SET)
+                )
+            );
+            ExecutionException exception = assertThrows(
+                ExecutionException.class,
+                () -> admin.incrementalAlterConfigs(alterations).all().get(1, TimeUnit.MINUTES)
+            );
+            assertEquals(
+                "org.apache.kafka.common.errors.InvalidRequestException: " +
+                    "Null value not supported for : my.custom.config, my.custom.config1",
+                exception.getMessage()
+            );
+        }
+    }
 }