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 (suggestion):
Instead of 

`kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()` ->

`kafkaConfig.isRemoteLogStorageSystemEnabled`

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 = {