/**
  * 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 kafka.server.logger.RuntimeLoggerManager

import java.util
import java.util.Properties
import kafka.server.metadata.ConfigRepository
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.ConfigDef.ConfigKey
import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, CLIENT_METRICS, TOPIC}
import org.apache.kafka.common.config.{ConfigDef, ConfigResource}
import org.apache.kafka.common.errors.{ApiException, 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}
import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.{AlterConfigsResourceResponse => IAlterConfigsResourceResponse}
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 org.slf4j.{Logger, LoggerFactory}

import scala.collection.{Map, Seq}
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
 * [[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
 * [[org.apache.kafka.server.policy.AlterConfigPolicy]].
 */
class ConfigAdminManager(nodeId: Int,
                         conf: KafkaConfig,
                         configRepository: ConfigRepository) extends Logging {
  import ConfigAdminManager._

  this.logIdent = "[ConfigAdminManager[nodeId=" + nodeId + "]: "

  val runtimeLoggerManager = new RuntimeLoggerManager(nodeId, logger.underlying)

  /**
   * Preprocess an incremental configuration operation on the broker. This step handles
   * setting log4j levels, as well as filtering out some invalid resource requests that
   * should not be forwarded to the controller.
   *
   * @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. The arguments
   *                    are the type and name of the resource to be authorized.
   *
   * @return            A map from resources to errors. If a resource appears in this map,
   *                    it has been preprocessed and does not need further processing.
   */
  def preprocess(
    request: IncrementalAlterConfigsRequestData,
    authorize: (ResourceType, String) => Boolean
  ): util.IdentityHashMap[IAlterConfigsResource, ApiError] = {
    val results = new util.IdentityHashMap[IAlterConfigsResource, ApiError]()
    val resourceIds = new util.HashMap[(Byte, String), IAlterConfigsResource]
    request.resources().forEach(resource => {
      val preexisting = resourceIds.put((resource.resourceType(), resource.resourceName()), resource)
      if (preexisting != null) {
        Seq(preexisting, resource).foreach(
          r => results.put(r, new ApiError(INVALID_REQUEST, "Each resource must appear at most once.")))
      }
    })
    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())
            }
          }
          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 =>
            // Nothing to do.
            case _ =>
              throw new InvalidRequestException(s"Unknown resource type ${resource.resourceType().toInt}")
          }
        } catch {
          case t: Throwable =>
            val err = ApiError.fromThrowable(t)
            info(s"Error preprocessing incrementalAlterConfigs request on $configResource", t)
            results.put(resource, err)
        }
      }
    })
    results
  }

  private def validateBrokerConfigChange(
    resource: IAlterConfigsResource,
    configResource: ConfigResource
  ): Unit = {
    val perBrokerConfig = configResource.name().nonEmpty
    val persistentProps = configRepository.config(configResource)
    val configProps = conf.dynamicConfig.fromPersistentProps(persistentProps, perBrokerConfig)
    val alterConfigOps = resource.configs().asScala.map {
      config =>
        val opType = AlterConfigOp.OpType.forId(config.configOperation())
        if (opType == null) {
          throw new InvalidRequestException(s"Unknown operations type ${config.configOperation}")
        }
        new AlterConfigOp(new ConfigEntry(config.name(), config.value()), opType)
    }.toSeq
    prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
    try {
      validateBrokerConfigChange(configProps, configResource)
    } catch {
      case t: Throwable => error(s"validation of configProps $configProps for $configResource failed with exception", t)
        throw t
    }
  }

  private def validateBrokerConfigChange(
    props: Properties,
    configResource: ConfigResource
  ): Unit = {
    try {
      conf.dynamicConfig.validate(props, configResource.name().nonEmpty)
    } catch {
      case e: ApiException => throw e
      //KAFKA-13609: InvalidRequestException is not really the right exception here if the
      // configuration fails validation. The configuration is still well-formed, but just
      // can't be applied. It should probably throw InvalidConfigurationException. However,
      // we should probably only change this in a KIP since it has compatibility implications.
      case e: Throwable => throw new InvalidRequestException(e.getMessage)
    }
 }

  /**
   * Preprocess a legacy configuration operation on the broker.
   *
   * @param request     The request data.
   *
   * @return
   */
  def preprocess(
    request: AlterConfigsRequestData,
  ): util.IdentityHashMap[LAlterConfigsResource, ApiError] = {
    val results = new util.IdentityHashMap[LAlterConfigsResource, ApiError]()
    val resourceIds = new util.HashMap[(Byte, String), LAlterConfigsResource]
    request.resources().forEach(resource => {
      val preexisting = resourceIds.put((resource.resourceType(), resource.resourceName()), resource)
      if (preexisting != null) {
        Seq(preexisting, resource).foreach(
          r => results.put(r, new ApiError(INVALID_REQUEST, "Each resource must appear at most once.")))
      }
    })
    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.value() == null) {
              nullUpdates.add(config.name())
            }
          }
          if (!nullUpdates.isEmpty) {
            throw new InvalidRequestException("Null value not supported for : " +
              String.join(", ", nullUpdates))
          }
          resourceType match {
            case BROKER =>
              if (configResource.name().nonEmpty) {
                validateResourceNameIsCurrentNodeId(resource.resourceName())
              }
              validateBrokerConfigChange(resource, configResource)
            case TOPIC | CLIENT_METRICS =>
            // Nothing to do.
            case _ =>
              // Since legacy AlterConfigs does not support BROKER_LOGGER, any attempt to use it
              // gets caught by this clause.
              throw new InvalidRequestException(s"Unknown resource type ${resource.resourceType().toInt}")
          }
        } catch {
          case t: Throwable =>
            val err = ApiError.fromThrowable(t)
            info(s"Error preprocessing alterConfigs request on $configResource: $err")
            results.put(resource, err)
        }
      }
    })
    results
  }

  private def validateBrokerConfigChange(
    resource: LAlterConfigsResource,
    configResource: ConfigResource
  ): Unit = {
    val props = new Properties()
    resource.configs().forEach {
      config => props.setProperty(config.name(), config.value())
    }
    validateBrokerConfigChange(props, configResource)
  }

  def validateResourceNameIsCurrentNodeId(name: String): Unit = {
    val id = try name.toInt catch {
      case _: NumberFormatException =>
        throw new InvalidRequestException(s"Node id must be an integer, but it is: $name")
    }
    if (id != nodeId) {
      throw new InvalidRequestException(s"Unexpected broker id, expected $nodeId, but received $name")
    }
  }
}

object ConfigAdminManager {
  val log: Logger = LoggerFactory.getLogger(classOf[ConfigAdminManager])

  /**
   * Copy the incremental configs request data without any already-processed elements.
   *
   * @param request   The input request. Will not be modified.
   * @param processed A map containing the resources that have already been processed.
   * @return          A new request object.
   */
  def copyWithoutPreprocessed(
    request: IncrementalAlterConfigsRequestData,
    processed: util.IdentityHashMap[IAlterConfigsResource, ApiError]
  ): IncrementalAlterConfigsRequestData = {
    val copy = new IncrementalAlterConfigsRequestData().
      setValidateOnly(request.validateOnly())
    request.resources().forEach(resource => {
      if (!processed.containsKey(resource)) {
        copy.resources().mustAdd(resource.duplicate())
      }
    })
    copy
  }

  /**
   * Copy the legacy alter configs request data without any already-processed elements.
   *
   * @param request   The input request. Will not be modified.
   * @param processed A map containing the resources that have already been processed.
   * @return          A new request object.
   */
  def copyWithoutPreprocessed(
    request: AlterConfigsRequestData,
    processed: util.IdentityHashMap[LAlterConfigsResource, ApiError]
  ): AlterConfigsRequestData = {
    val copy = new AlterConfigsRequestData().
      setValidateOnly(request.validateOnly())
    request.resources().forEach(resource => {
      if (!processed.containsKey(resource)) {
        copy.resources().mustAdd(resource.duplicate())
      }
    })
    copy
  }

  def reassembleIncrementalResponse(
    original: IncrementalAlterConfigsRequestData,
    preprocessingResponses: util.IdentityHashMap[IAlterConfigsResource, ApiError],
    persistentResponses: IncrementalAlterConfigsResponseData
  ): IncrementalAlterConfigsResponseData = {
    val response = new IncrementalAlterConfigsResponseData()
    val responsesByResource = persistentResponses.responses().iterator().asScala.map(r => (r.resourceName(), r.resourceType()) -> new ApiError(r.errorCode(), r.errorMessage())).toMap
    original.resources().forEach(r => {
      val err = Option(preprocessingResponses.get(r)) match {
        case None =>
          responsesByResource.get((r.resourceName(), r.resourceType())) match {
            case None => log.error("The controller returned fewer results than we " +
              s"expected. No response found for $r.")
              new ApiError(UNKNOWN_SERVER_ERROR)
            case Some(err) => err
          }
        case Some(err) => err
      }
      response.responses().add(new IAlterConfigsResourceResponse().
        setResourceName(r.resourceName()).
        setResourceType(r.resourceType()).
        setErrorCode(err.error().code()).
        setErrorMessage(err.message()))
    })
    response
  }

  def reassembleLegacyResponse(
    original: AlterConfigsRequestData,
    preprocessingResponses: util.IdentityHashMap[LAlterConfigsResource, ApiError],
    persistentResponses: AlterConfigsResponseData
  ): AlterConfigsResponseData = {
    val response = new AlterConfigsResponseData()
    val responsesByResource = persistentResponses.responses().iterator().asScala.map(r => (r.resourceName(), r.resourceType()) -> new ApiError(r.errorCode(), r.errorMessage())).toMap
    original.resources().forEach(r => {
      val err = Option(preprocessingResponses.get(r)) match {
        case None =>
          responsesByResource.get((r.resourceName(), r.resourceType())) match {
            case None => log.error("The controller returned fewer results than we " +
              s"expected. No response found for $r.")
              new ApiError(UNKNOWN_SERVER_ERROR)
            case Some(err) => err
          }
        case Some(err) => err
      }
      response.responses().add(new LAlterConfigsResourceResponse().
        setResourceName(r.resourceName()).
        setResourceType(r.resourceType()).
        setErrorCode(err.error().code()).
        setErrorMessage(err.message()))
    })
    response
  }

  def containsDuplicates[T](
    iterable: Iterable[T]
  ): Boolean = {
    val previous = new util.HashSet[T]()
    !iterable.forall(previous.add)
  }

  /**
   * Convert the configuration properties for an object (broker, topic, etc.) to a Scala
   * map. Sensitive configurations will be redacted, so that the output is suitable for
   * logging.
   *
   * @param resource      The configuration resource.
   * @param configProps   The configuration as a Properties object.
   * @return              A map containing all the configuration keys and values, as they
   *                      should be logged.
   */
  def toLoggableProps(resource: ConfigResource, configProps: Properties): Map[String, String] = {
    configProps.asScala.map {
      case (key, value) => (key, KafkaConfig.loggableValue(resource.`type`, key, value))
    }
  }

  /**
   * Apply a series of incremental configuration operations to a set of resource properties.
   *
   * @param alterConfigOps  The incremental configuration operations to apply.
   * @param configProps     The resource properties. This will be modified by this function.
   * @param configKeys      Information about configuration key types.
   */
  def prepareIncrementalConfigs(
    alterConfigOps: Seq[AlterConfigOp],
    configProps: Properties,
    configKeys: Map[String, ConfigKey]
  ): Unit = {
    def listType(configName: String, configKeys: Map[String, ConfigKey]): Boolean = {
      val configKey = configKeys(configName)
      if (configKey == null)
        throw new InvalidConfigurationException(s"Unknown config name: $configName")
      configKey.`type` == ConfigDef.Type.LIST
    }

    alterConfigOps.foreach { alterConfigOp =>
      val configPropName = alterConfigOp.configEntry.name
      alterConfigOp.opType() match {
        case OpType.SET => configProps.setProperty(alterConfigOp.configEntry.name, alterConfigOp.configEntry.value)
        case OpType.DELETE => configProps.remove(alterConfigOp.configEntry.name)
        case OpType.APPEND =>
          if (!listType(alterConfigOp.configEntry.name, configKeys))
            throw new InvalidConfigurationException(s"Config value append is not allowed for config key: ${alterConfigOp.configEntry.name}")
          val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
            .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
            .filter(s => s.nonEmpty)
            .map(_.split(",").toList)
            .getOrElse(List.empty)
          val appendingValueList = alterConfigOp.configEntry.value.split(",").toList.filter(value => !oldValueList.contains(value))
          val newValueList = oldValueList ::: appendingValueList
          configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
        case OpType.SUBTRACT =>
          if (!listType(alterConfigOp.configEntry.name, configKeys))
            throw new InvalidConfigurationException(s"Config value subtract is not allowed for config key: ${alterConfigOp.configEntry.name}")
          val oldValueList = Option(configProps.getProperty(alterConfigOp.configEntry.name))
            .orElse(Option(ConfigDef.convertToString(configKeys(configPropName).defaultValue, ConfigDef.Type.LIST)))
            .getOrElse("")
            .split(",").toList
          val newValueList = oldValueList.diff(alterConfigOp.configEntry.value.split(",").toList)
          configProps.setProperty(alterConfigOp.configEntry.name, newValueList.mkString(","))
      }
    }
  }
}
