[GitHub] [kafka] cmccabe commented on a change in pull request #11657: KAFKA-13552: Fix BROKER and BROKER_LOGGER in KRaft

[GitBox <gi...@apache.org>]

cmccabe commented on a change in pull request #11657:
URL: https://github.com/apache/kafka/pull/11657#discussion_r784328261



##########
File path: core/src/main/scala/kafka/server/ConfigAdminManager.scala
##########
@@ -0,0 +1,373 @@
+/**
+  * Licensed to the Apache Software Foundation (ASF) under one or more
+  * contributor license agreements.  See the NOTICE file distributed with
+  * this work for additional information regarding copyright ownership.
+  * The ASF licenses this file to You under the Apache License, Version 2.0
+  * (the "License"); you may not use this file except in compliance with
+  * the License.  You may obtain a copy of the License at
+  *
+  *    http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+package kafka.server
+
+import java.util
+import java.util.Properties
+
+import kafka.log.LogConfig
+import kafka.server.DynamicConfigManager.prepareIncrementalConfigs
+import kafka.utils.Log4jController
+import kafka.utils._
+import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry}
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, TOPIC}
+import org.apache.kafka.common.config.{ConfigResource, LogLevelConfig}
+import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidConfigurationException, InvalidRequestException}
+import org.apache.kafka.common.message.{AlterConfigsRequestData, AlterConfigsResponseData, IncrementalAlterConfigsRequestData, IncrementalAlterConfigsResponseData}
+import org.apache.kafka.common.message.AlterConfigsRequestData.{AlterConfigsResource => LAlterConfigsResource}
+import org.apache.kafka.common.message.AlterConfigsResponseData.{AlterConfigsResourceResponse => LAlterConfigsResourceResponse}
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.{AlterConfigsResource => IAlterConfigsResource, AlterableConfig => IAlterableConfig}
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.{AlterConfigsResourceResponse => IAlterConfigsResourceResponse}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.protocol.Errors.{INVALID_REQUEST, UNKNOWN_SERVER_ERROR}
+import org.apache.kafka.common.requests.ApiError
+import org.apache.kafka.common.resource.{Resource, ResourceType}
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * Manages dynamic configuration operations on the broker.
+ *
+ * There are two RPCs that alter KIP-226 dynamic configurations: alterConfigs, and
+ * incrementalAlterConfigs. The main difference between the two is that alterConfigs sets
+ * all configurations related to a specific config resource, whereas
+ * incrementalAlterConfigs makes only designated changes.
+ *
+ * The original, non-incremental AlterConfigs is deprecated because there are inherent
+ * race conditions when multiple clients use it. It deletes any resource configuration
+ * keys that are not specified. This leads to clients trying to do read-modify-write
+ * cycles when they only want to change one config key. (But even read-modify-write doesn't
+ * work correctly, since "sensitive" configurations are omitted when read.)
+ *
+ * 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 (in ZK or KRaft) 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.
+ * - The first step, called "preprocessing," handles setting KIP-412 log levels, validating
+ * BROKER configurations. We also filter out some other things here like UNKNOWN resource
+ * types, etc.
+ * - The second step is "persistence," and handles storing the configurations durably to our
+ * metadata store.
+ *
+ * When KIP-590 forwarding is active (such as in KRaft mode), preprocessing will happen
+ * on the broker, while persistence will happen on the active controller. (If KIP-590
+ * forwarding is not active, then both steps are done on the same broker.)
+ *
+ * In KRaft mode, the active controller performs its own configuration validation step in
+ * {@link kafka.server.ControllerConfigurationValidator}. This is mainly important for
+ * TOPIC resources, since we already validated changes to BROKER resources on the
+ * forwarding broker. The KRaft controller is also responsible for enforcing the configured
+ * {@link org.apache.kafka.server.policy.AlterConfigPolicy}.
+ */
+class ConfigAdminManager(nodeId: Int,
+                         conf: KafkaConfig,
+                         getResourceConfig: ConfigResource => Properties) extends Logging {
+  this.logIdent = "[ConfigAdminManager[nodeId=" + nodeId + "]: "
+
+  /**
+   * Preprocess an incremental configuration operation on the broker.
+   *
+   * @param request     The request data.
+   * @param authorize   A callback which is invoked when we need to authorize an operation.
+   *                    Currently, we only use this for log4j operations. Other types of
+   *                    operations are authorized in the persistence step.

Review comment:
       Added




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org