Repository: apache/kafka
File: core/src/main/scala/kafka/server/ConfigHandler.scala (Scala)

Code before:

/**
  * The ConfigHandler is used to process broker configuration change notifications.
  */
trait ConfigHandler {
  def processConfigChanges(entityName: String, value: Properties): Unit
}

/**
  * The TopicConfigHandler will process topic config changes from ZooKeeper or the metadata log.
  * The callback provides the topic name and the full properties set.
  */
class TopicConfigHandler(private val replicaManager: ReplicaManager,
                         kafkaConfig: KafkaConfig,
                         val quotas: QuotaManagers,
                         kafkaController: Option[KafkaController]) extends ConfigHandler with Logging  {

  private def updateLogConfig(topic: String,
                              topicConfig: Properties): Unit = {
    val logManager = replicaManager.logManager
    // Validate the configurations.
    val configNamesToExclude = excludedConfigs(topic, topicConfig)
    val props = new Properties()
    topicConfig.asScala.forKeyValue { (key, value) =>
      if (!configNamesToExclude.contains(key)) props.put(key, value)
    }
    val logs = logManager.logsByTopic(topic)
    val wasRemoteLogEnabledBeforeUpdate = logs.exists(_.remoteLogEnabled())
    logManager.updateTopicConfig(topic, props)
    maybeBootstrapRemoteLogComponents(topic, logs, wasRemoteLogEnabledBeforeUpdate)
  }

  private[server] def maybeBootstrapRemoteLogComponents(topic: String,
                                                        logs: Seq[UnifiedLog],
                                                        wasRemoteLogEnabledBeforeUpdate: Boolean): Unit = {
    val isRemoteLogEnabled = logs.exists(_.remoteLogEnabled())
    // Topic configs gets updated incrementally. This check is added to prevent redundant updates.
    if (!wasRemoteLogEnabledBeforeUpdate && isRemoteLogEnabled) {
      val (leaderPartitions, followerPartitions) =
        logs.flatMap(log => replicaManager.onlinePartition(log.topicPartition)).partition(_.isLeader)
      val topicIds = Collections.singletonMap(topic, replicaManager.metadataCache.getTopicId(topic))
      replicaManager.remoteLogManager.foreach(rlm =>
        rlm.onLeadershipChange(leaderPartitions.toSet.asJava, followerPartitions.toSet.asJava, topicIds))
    } else if (wasRemoteLogEnabledBeforeUpdate && !isRemoteLogEnabled) {
      warn(s"Disabling remote log on the topic: $topic is not supported.")
    }
  }

  def processConfigChanges(topic: String, topicConfig: Properties): Unit = {
    updateLogConfig(topic, topicConfig)


Review comment (bug):
Correct me if I am wrong here but this method is called when each broker is trying to implement the configuration change sent to it by metadata topic (kraft) or zk. Even if we reject it here, the config change will continue to persist in Zk and metadata.

We need to block this configuration change at the start itself, i.e. in KafkaConfig.scala 

Code after:

/**
  * The ConfigHandler is used to process broker configuration change notifications.
  */
trait ConfigHandler {
  def processConfigChanges(entityName: String, value: Properties): Unit
}

/**
  * The TopicConfigHandler will process topic config changes from ZooKeeper or the metadata log.
  * The callback provides the topic name and the full properties set.
  */
class TopicConfigHandler(private val replicaManager: ReplicaManager,
                         kafkaConfig: KafkaConfig,
                         val quotas: QuotaManagers,
                         kafkaController: Option[KafkaController]) extends ConfigHandler with Logging  {

  private def updateLogConfig(topic: String,
                              topicConfig: Properties): Unit = {
    val logManager = replicaManager.logManager
    // Validate the configurations.
    val configNamesToExclude = excludedConfigs(topic, topicConfig)
    val props = new Properties()
    topicConfig.asScala.forKeyValue { (key, value) =>
      if (!configNamesToExclude.contains(key)) props.put(key, value)
    }

    val logs = logManager.logsByTopic(topic)
    val wasRemoteLogEnabledBeforeUpdate = logs.exists(_.remoteLogEnabled())

    logManager.updateTopicConfig(topic, props, kafkaConfig.isRemoteLogStorageSystemEnabled)
    maybeBootstrapRemoteLogComponents(topic, logs, wasRemoteLogEnabledBeforeUpdate)
  }

  private[server] def maybeBootstrapRemoteLogComponents(topic: String,
                                                        logs: Seq[UnifiedLog],
                                                        wasRemoteLogEnabledBeforeUpdate: Boolean): Unit = {
    val isRemoteLogEnabled = logs.exists(_.remoteLogEnabled())
    // Topic configs gets updated incrementally. This check is added to prevent redundant updates.
    if (!wasRemoteLogEnabledBeforeUpdate && isRemoteLogEnabled) {
      val (leaderPartitions, followerPartitions) =
        logs.flatMap(log => replicaManager.onlinePartition(log.topicPartition)).partition(_.isLeader)
      val topicIds = Collections.singletonMap(topic, replicaManager.metadataCache.getTopicId(topic))
      replicaManager.remoteLogManager.foreach(rlm =>
        rlm.onLeadershipChange(leaderPartitions.toSet.asJava, followerPartitions.toSet.asJava, topicIds))
    } else if (wasRemoteLogEnabledBeforeUpdate && !isRemoteLogEnabled) {
      warn(s"Disabling remote log on the topic: $topic is not supported.")
    }
  }

  def processConfigChanges(topic: String, topicConfig: Properties): Unit = {