/**
 * This is the zookeeper listener that triggers all the state transitions for a replica
  * 唯一的Zookeeper Listener，监听/brokers/ids节点下的子节点变化，主要负责处理Broker的上线和故障下线。
  * 当Broker上线时会在/brokers/ids下创建临时节点，下线时会删除对应的临时节点。
 */
class BrokerChangeListener() extends IZkChildListener with Logging {
  this.logIdent = "[BrokerChangeListener on Controller " + controller.config.brokerId + "]: "
  def handleChildChange(parentPath : String, currentBrokerList : java.util.List[String]) {
    info("Broker change listener fired for path %s with children %s".format(parentPath, currentBrokerList.sorted.mkString(",")))
    inLock(controllerContext.controllerLock) {
      if (hasStarted.get) { // 需要在状态机启动的情况下执行
        ControllerStats.leaderElectionTimer.time { // 计时操作
          try {
            // 获取Zookeeper中的Broker列表、得到BrokerID集合
            val curBrokers = currentBrokerList.map(_.toInt).toSet.flatMap(zkUtils.getBrokerInfo)
            val curBrokerIds = curBrokers.map(_.id)
            // ControllerContext中记录的当前可用Broker的ID集合
            val liveOrShuttingDownBrokerIds = controllerContext.liveOrShuttingDownBrokerIds
            // 过滤得到新增的Broker列表
            val newBrokerIds = curBrokerIds -- liveOrShuttingDownBrokerIds
            // 过滤得到出现故障的Broker的ID集合
            val deadBrokerIds = liveOrShuttingDownBrokerIds -- curBrokerIds
            val newBrokers = curBrokers.filter(broker => newBrokerIds(broker.id))
            // 更新ControllerContext中记录的可用Broker列表
            controllerContext.liveBrokers = curBrokers

            // 对新增Broker的列表、出现故障Broker的列表以及Zookeeper中的Broker列表进行排序
            val newBrokerIdsSorted = newBrokerIds.toSeq.sorted
            val deadBrokerIdsSorted = deadBrokerIds.toSeq.sorted
            val liveBrokerIdsSorted = curBrokerIds.toSeq.sorted
            info("Newly added brokers: %s, deleted brokers: %s, all live brokers: %s"
              .format(newBrokerIdsSorted.mkString(","), deadBrokerIdsSorted.mkString(","), liveBrokerIdsSorted.mkString(",")))

            // 创建Controller和新增Broker的网络连接
            newBrokers.foreach(controllerContext.controllerChannelManager.addBroker)
            // 关闭Controller和出现故障的Broker的网络连接
            deadBrokerIds.foreach(controllerContext.controllerChannelManager.removeBroker)

            // 调用onBrokerStartup()方法和onBrokerFailure()方法分别处理新增Broker和下线Broker
            if(newBrokerIds.size > 0)
              controller.onBrokerStartup(newBrokerIdsSorted)
            if(deadBrokerIds.size > 0)
              controller.onBrokerFailure(deadBrokerIdsSorted)
          } catch {
            case e: Throwable => error("Error while handling broker changes", e)
          }
        }
      }
    }
  }
}

/**
 * This callback is invoked by the replica state machine's broker change listener, with the list of newly started
 * brokers as input. It does the following -
 * 1. Sends update metadata request to all live and shutting down brokers
 * 2. Triggers the OnlinePartition state change for all new/offline partitions
 * 3. It checks whether there are reassigned replicas assigned to any newly started brokers.  If
 *    so, it performs the reassignment logic for each topic/partition.
 *
 * Note that we don't need to refresh the leader/isr cache for all topic/partitions at this point for two reasons:
 * 1. The partition state machine, when triggering online state change, will refresh leader and ISR for only those
 *    partitions currently new or offline (rather than every partition this controller is aware of)
 * 2. Even if we do refresh the cache, there is no guarantee that by the time the leader and ISR request reaches
 *    every broker that it is still valid.  Brokers check the leader epoch to determine validity of the request.
 */
def onBrokerStartup(newBrokers: Seq[Int]) {
  info("New broker startup callback for %s".format(newBrokers.mkString(",")))
  val newBrokersSet = newBrokers.toSet
  // send update metadata request to all live and shutting down brokers. Old brokers will get to know of the new
  // broker via this update.
  // In cases of controlled shutdown leaders will not be elected when a new broker comes up. So at least in the
  // common controlled shutdown case, the metadata will reach the new brokers faster
  /**
    * 向集群中所有Broker发送UpdateMetadataRequest，发送的是所有可用的Broker的ID信息
    * 通过此请求，集群中所有Broker可以了解到新添加的Broker信息
    */
  sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
  // the very first thing to do when a new broker comes up is send it the entire list of partitions that it is
  // supposed to host. Based on that the broker starts the high watermark threads for the input list of partitions
  // 将新增Broker上的副本转换为OnlineReplica状态，此步骤会涉及发送LeaderAndIsrRequest
  val allReplicasOnNewBrokers = controllerContext.replicasOnBrokers(newBrokersSet)
  replicaStateMachine.handleStateChanges(allReplicasOnNewBrokers, OnlineReplica)
  // when a new broker comes up, the controller needs to trigger leader election for all new and offline partitions
  // to see if these brokers can become leaders for some/all of those
  // 将NewPartition和OfflinePartition状态的分区转换为OnlinePartition状态
  partitionStateMachine.triggerOnlinePartitionStateChange()
  // check if reassignment of some partitions need to be restarted
  // 检测进行副本重新分配
  val partitionsWithReplicasOnNewBrokers = controllerContext.partitionsBeingReassigned.filter {
    case (topicAndPartition, reassignmentContext) => reassignmentContext.newReplicas.exists(newBrokersSet.contains(_))
  }
  partitionsWithReplicasOnNewBrokers.foreach(p => onPartitionReassignment(p._1, p._2))
  // check if topic deletion needs to be resumed. If at least one replica that belongs to the topic being deleted exists
  // on the newly restarted brokers, there is a chance that topic deletion can resume
  // 如果新增Broker上有待删除的Topic的副本，则唤醒DeleteTopicsThread线程进行删除
  val replicasForTopicsToBeDeleted = allReplicasOnNewBrokers.filter(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
  if(replicasForTopicsToBeDeleted.size > 0) {
    info(("Some replicas %s for topics scheduled for deletion %s are on the newly restarted brokers %s. " +
      "Signaling restart of topic deletion for these topics").format(replicasForTopicsToBeDeleted.mkString(","),
      deleteTopicManager.topicsToBeDeleted.mkString(","), newBrokers.mkString(",")))
    deleteTopicManager.resumeDeletionForTopics(replicasForTopicsToBeDeleted.map(_.topic))
  }
}

/**
 * This callback is invoked by the replica state machine's broker change listener with the list of failed brokers
 * as input. It does the following -
 * 1. Mark partitions with dead leaders as offline
 * 2. Triggers the OnlinePartition state change for all new/offline partitions
 * 3. Invokes the OfflineReplica state change on the input list of newly started brokers
 * 4. If no partitions are effected then send UpdateMetadataRequest to live or shutting down brokers
 *
 * Note that we don't need to refresh the leader/isr cache for all topic/partitions at this point.  This is because
 * the partition state machine will refresh our cache for us when performing leader election for all new/offline
 * partitions coming online.
 */
def onBrokerFailure(deadBrokers: Seq[Int]) {
  info("Broker failure callback for %s".format(deadBrokers.mkString(",")))
  // 将正在正常关闭的Broker从deadbroker列表中移除
  val deadBrokersThatWereShuttingDown =
    deadBrokers.filter(id => controllerContext.shuttingDownBrokerIds.remove(id))
  info("Removed %s from list of shutting down brokers.".format(deadBrokersThatWereShuttingDown))
  val deadBrokersSet = deadBrokers.toSet
  // trigger OfflinePartition state for all partitions whose current leader is one amongst the dead brokers
  // 过滤得到Leader副本在故障Broker上的分区，将其转换为OfflinePartition状态
  val partitionsWithoutLeader = controllerContext.partitionLeadershipInfo.filter(partitionAndLeader =>
    deadBrokersSet.contains(partitionAndLeader._2.leaderAndIsr.leader) &&
      !deleteTopicManager.isTopicQueuedUpForDeletion(partitionAndLeader._1.topic)).keySet
  partitionStateMachine.handleStateChanges(partitionsWithoutLeader, OfflinePartition)
  // trigger OnlinePartition state changes for offline or new partitions
  // 将OfflinePartition状态的分区转换为OnlinePartition状态
  partitionStateMachine.triggerOnlinePartitionStateChange()
  // filter out the replicas that belong to topics that are being deleted
  // 过滤得到在故障Broker上的副本，将这些副本转换为OfflinePartition状态
  var allReplicasOnDeadBrokers = controllerContext.replicasOnBrokers(deadBrokersSet)
  val activeReplicasOnDeadBrokers = allReplicasOnDeadBrokers.filterNot(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
  // handle dead replicas
  replicaStateMachine.handleStateChanges(activeReplicasOnDeadBrokers, OfflineReplica)
  // check if topic deletion state for the dead replicas needs to be updated
  // 检查故障Broker上是否有待删除Topic的副本，如果存在则将其转换为ReplicaDeletionIneligible状态并标记Topic不可删除
  val replicasForTopicsToBeDeleted = allReplicasOnDeadBrokers.filter(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
  if(replicasForTopicsToBeDeleted.size > 0) {
    // it is required to mark the respective replicas in TopicDeletionFailed state since the replica cannot be
    // deleted when the broker is down. This will prevent the replica from being in TopicDeletionStarted state indefinitely
    // since topic deletion cannot be retried until at least one replica is in TopicDeletionStarted state
    deleteTopicManager.failReplicaDeletion(replicasForTopicsToBeDeleted)
  }

  // If broker failure did not require leader re-election, inform brokers of failed broker
  // Note that during leader re-election, brokers update their metadata
  // 发送UpdateMetadataRequest更新所有Broker的信息
  if (partitionsWithoutLeader.isEmpty) {
    sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
  }
}

def main(args: Array[String]): Unit = {
  try {
    val serverProps = getPropsFromArgs(args)
    val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)

    // attach shutdown handler to catch control-c
    // 虚拟机钩子
    Runtime.getRuntime().addShutdownHook(new Thread() {
      override def run() = {
        // 关闭KafkaController，该方法会调用controlledShutdown()方法向Controller Leader发送ControlledShutdownRequest
        kafkaServerStartable.shutdown
      }
    })

    kafkaServerStartable.startup
    kafkaServerStartable.awaitShutdown
  }
  catch {
    case e: Throwable =>
      fatal(e)
      System.exit(1)
  }
  System.exit(0)
}

/**
 * On clean shutdown, the controller first determines the partitions that the
 * shutting down broker leads, and moves leadership of those partitions to another broker
 * that is in that partition's ISR.
 *
 * @param id Id of the broker to shutdown.
 * @return The number of partitions that the broker still leads.
 */
def shutdownBroker(id: Int) : Set[TopicAndPartition] = {

  // 检查Controller是否存活
  if (!isActive()) {
    throw new ControllerMovedException("Controller moved to another broker. Aborting controlled shutdown")
  }

  controllerContext.brokerShutdownLock synchronized { // 加锁
    info("Shutting down broker " + id)

    inLock(controllerContext.controllerLock) {
      // 检查Broker是否存活
      if (!controllerContext.liveOrShuttingDownBrokerIds.contains(id))
        throw new BrokerNotAvailableException("Broker id %d does not exist.".format(id))

      // 将BrokerID添加到ControllerContext的shuttingDownBrokerIds中
      controllerContext.shuttingDownBrokerIds.add(id)
      debug("All shutting down brokers: " + controllerContext.shuttingDownBrokerIds.mkString(","))
      debug("Live brokers: " + controllerContext.liveBrokerIds.mkString(","))
    }

    // 获取待关闭Broker上所有的Partition和副本信息
    val allPartitionsAndReplicationFactorOnBroker: Set[(TopicAndPartition, Int)] =
      inLock(controllerContext.controllerLock) {
        controllerContext.partitionsOnBroker(id) // 所有Partition
          .map(topicAndPartition => (topicAndPartition, controllerContext.partitionReplicaAssignment(topicAndPartition).size))
      }

    allPartitionsAndReplicationFactorOnBroker.foreach { // 遍历待关闭的Broker上的分区
      case(topicAndPartition, replicationFactor) =>
        // Move leadership serially to relinquish lock.
        inLock(controllerContext.controllerLock) {
          // 获取分区的Leader副本所在Broker的ID、ISR集合、年代信息
          controllerContext.partitionLeadershipInfo.get(topicAndPartition).foreach { currLeaderIsrAndControllerEpoch =>
            if (replicationFactor > 1) { // 检查是否开启副本机制
              if (currLeaderIsrAndControllerEpoch.leaderAndIsr.leader == id) {
                // If the broker leads the topic partition, transition the leader and update isr. Updates zk and
                // notifies all affected brokers
                /**
                  * 将相关的分区切换为OnlinePartition状态，使用ControlledShutdownLeaderSelector重新选择Leader和ISR集合
                  * 并将结果写入ZooKeeper，之后发送LeaderAndIsrRequest和UpdateMetadataRequest
                  */
                partitionStateMachine.handleStateChanges(Set(topicAndPartition), OnlinePartition,
                  controlledShutdownPartitionLeaderSelector)
              } else {
                // Stop the replica first. The state change below initiates ZK changes which should take some time
                // before which the stop replica request should be completed (in most cases)
                try {
                  // 发送StopReplicaRequest（不删除副本）
                  brokerRequestBatch.newBatch()
                  brokerRequestBatch.addStopReplicaRequestForBrokers(Seq(id), topicAndPartition.topic,
                    topicAndPartition.partition, deletePartition = false)
                  brokerRequestBatch.sendRequestsToBrokers(epoch)
                } catch {
                  case e : IllegalStateException => {
                    // Resign if the controller is in an illegal state
                    error("Forcing the controller to resign")
                    brokerRequestBatch.clear()
                    controllerElector.resign()

                    throw e
                  }
                }
                // If the broker is a follower, updates the isr in ZK and notifies the current leader
                // 将副本转换为OfflineReplica状态
                replicaStateMachine.handleStateChanges(Set(PartitionAndReplica(topicAndPartition.topic,
                  topicAndPartition.partition, id)), OfflineReplica)
              }
            }
          }
        }
    }

    // 该方法用于统计Leader副本依然处于待关闭Broker上的分区
    def replicatedPartitionsBrokerLeads() = inLock(controllerContext.controllerLock) {
      trace("All leaders = " + controllerContext.partitionLeadershipInfo.mkString(","))
      controllerContext.partitionLeadershipInfo.filter {
        case (topicAndPartition, leaderIsrAndControllerEpoch) =>
          leaderIsrAndControllerEpoch.leaderAndIsr.leader == id && controllerContext.partitionReplicaAssignment(topicAndPartition).size > 1
      }.map(_._1)
    }

    // 统计Leader副本依然处于待关闭Broker上的分区，转换为Set后返回
    replicatedPartitionsBrokerLeads().toSet
  }
}

/**
 * This is the zookeeper listener that triggers all the state transitions for a partition
  * TopicChangeListener负责管理Topic的增删，它监听/brokers/topics节点的子节点的变化
 */
class TopicChangeListener extends IZkChildListener with Logging {
  this.logIdent = "[TopicChangeListener on Controller " + controller.config.brokerId + "]: "

  @throws(classOf[Exception])
  def handleChildChange(parentPath : String, children : java.util.List[String]) {
    inLock(controllerContext.controllerLock) {
      if (hasStarted.get) {
        try {
          val currentChildren = { // 获取/brokers/topics下的子节点集合
            import JavaConversions._
            debug("Topic change listener fired for path %s with children %s".format(parentPath, children.mkString(",")))
            (children: Buffer[String]).toSet
          }
          // 过滤，得到新添加的Topic
          val newTopics = currentChildren -- controllerContext.allTopics
          // 过滤，得到删除的Topic
          val deletedTopics = controllerContext.allTopics -- currentChildren
          // 更新ControllerContext的allTopics集合
          controllerContext.allTopics = currentChildren

          // 从Zookeeper的/brokers/topics/[topic_name]路径加载新增Topic的分区信息和AR集合信息
          val addedPartitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(newTopics.toSeq)
          // 更新ControllerContext的partitionReplicaAssignment记录的AR集合
          // 去掉删除主题的副本，添加新增主题的副本
          controllerContext.partitionReplicaAssignment = controllerContext.partitionReplicaAssignment.filter(p =>
            !deletedTopics.contains(p._1.topic))
          controllerContext.partitionReplicaAssignment.++=(addedPartitionReplicaAssignment)
          info("New topics: [%s], deleted topics: [%s], new partition replica assignment [%s]".format(newTopics,
            deletedTopics, addedPartitionReplicaAssignment))
          // 调用KafkaController的onNewTopicCreation()处理新增Topic
          if(newTopics.size > 0)
            controller.onNewTopicCreation(newTopics, addedPartitionReplicaAssignment.keySet.toSet)
        } catch {
          case e: Throwable => error("Error while handling new topic", e )
        }
      }
    }
  }
}

/**
 * This callback is invoked by the partition state machine's topic change listener with the list of new topics
 * and partitions as input. It does the following -
 * 1. Registers partition change listener. This is not required until KAFKA-347
 * 2. Invokes the new partition callback
 * 3. Send metadata request with the new topic to all brokers so they allow requests for that topic to be served
 */
def onNewTopicCreation(topics: Set[String], newPartitions: Set[TopicAndPartition]) {
  info("New topic creation callback for %s".format(newPartitions.mkString(",")))
  // subscribe to partition changes
  // 为每个新增的Topic注册PartitionModificationsListener监听器
  topics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic))
  // 完成新增Topic的分区状态及副本状态的转换
  onNewPartitionCreation(newPartitions)
}

/**
 * This callback is invoked by the topic change callback with the list of failed brokers as input.
 * It does the following -
 * 1. Move the newly created partitions to the NewPartition state
 * 2. Move the newly created partitions from NewPartition->OnlinePartition state
 */
def onNewPartitionCreation(newPartitions: Set[TopicAndPartition]) {
  info("New partition creation callback for %s".format(newPartitions.mkString(",")))
  // 将所有指定的新增分区转换为NewPartition状态
  partitionStateMachine.handleStateChanges(newPartitions, NewPartition)
  // 将指定分区的所有副本转换为NewReplica状态
  replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), NewReplica)
  // 将所有指定的新增分区转换为OnlinePartition状态
  partitionStateMachine.handleStateChanges(newPartitions, OnlinePartition, offlinePartitionSelector)
  // 将指定分区的所有副本转换为OnlineReplica状态
  replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), OnlineReplica)
}

/**
 * Delete topics includes the following operations -
 * 1. Add the topic to be deleted to the delete topics cache, only if the topic exists
 * 2. If there are topics to be deleted, it signals the delete topic thread
  *
  * 主题删除操作监听器
  * 1. 当被删除的主题存在时，将其添加到记录删除主题的缓存中；
  * 2. 当有主题被删除时，会通知DeleteTopicThread线程
 */
class DeleteTopicsListener() extends IZkChildListener with Logging {
  this.logIdent = "[DeleteTopicsListener on " + controller.config.brokerId + "]: "
  val zkUtils = controllerContext.zkUtils

  /**
   * Invoked when a topic is being deleted
   * @throws Exception On any error.
   */
  @throws(classOf[Exception])
  def handleChildChange(parentPath : String, children : java.util.List[String]) {
    inLock(controllerContext.controllerLock) {
      // 获取待删除的主题的集合
      var topicsToBeDeleted = {
        import JavaConversions._
        (children: Buffer[String]).toSet
      }
      debug("Delete topics listener fired for topics %s to be deleted".format(topicsToBeDeleted.mkString(",")))
      // 过滤得到不存在的删除主题
      val nonExistentTopics = topicsToBeDeleted.filter(t => !controllerContext.allTopics.contains(t))
      if(nonExistentTopics.size > 0) {
        warn("Ignoring request to delete non-existing topics " + nonExistentTopics.mkString(","))
        // 对于不存在的主题，直接将其在/admin/delete_topics下对应的节点删除
        nonExistentTopics.foreach(topic => zkUtils.deletePathRecursive(getDeleteTopicPath(topic)))
      }
      // 过滤掉不存在的待删除主题
      topicsToBeDeleted --= nonExistentTopics
      if(topicsToBeDeleted.size > 0) {
        info("Starting topic deletion for topics " + topicsToBeDeleted.mkString(","))
        // mark topic ineligible for deletion if other state changes are in progress
        topicsToBeDeleted.foreach { topic => // 遍历所有待删除的主题，检查待删除主题是否处于不可删除的情况
          // 第1项：检查待删除主题中是否有分区正在进行"优先副本"选举
          val preferredReplicaElectionInProgress =
            controllerContext.partitionsUndergoingPreferredReplicaElection.map(_.topic).contains(topic)
          // 第2项：检查待删除主题中是否有分区正在进行副本重新分配
          val partitionReassignmentInProgress =
            controllerContext.partitionsBeingReassigned.keySet.map(_.topic).contains(topic)
          if(preferredReplicaElectionInProgress || partitionReassignmentInProgress)
            // 有1项满足，就将待删除主题标记为不可删除
            controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))
        }
        // add topic to deletion list
        // 通过该方法将待删除的主题放入到topicsToBeDeleted集合，将待删除的主题的分区放入到partitionsToBeDeleted集合
        controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)
      }
    }
  }

  /**
   *
   * @throws Exception
 *             On any error.
   */
  @throws(classOf[Exception])
  def handleDataDeleted(dataPath: String) {
  }
}

class TopicDeletionManager(controller: KafkaController,
                           initialTopicsToBeDeleted: Set[String] = Set.empty,
                           initialTopicsIneligibleForDeletion: Set[String] = Set.empty) extends Logging {
  ...
  val controllerContext = controller.controllerContext
  // 分区状态机
  val partitionStateMachine = controller.partitionStateMachine
  // 副本状态机
  val replicaStateMachine = controller.replicaStateMachine
  // 记录将要被删除的主题集合
  val topicsToBeDeleted: mutable.Set[String] = mutable.Set.empty[String] ++ initialTopicsToBeDeleted
  // 记录将要被删除的分区集合
  val partitionsToBeDeleted: mutable.Set[TopicAndPartition] = topicsToBeDeleted.flatMap(controllerContext.partitionsForTopic)
  val deleteLock = new ReentrantLock()
  // 记录不可被删除的主题
  val topicsIneligibleForDeletion: mutable.Set[String] = mutable.Set.empty[String] ++
    (initialTopicsIneligibleForDeletion & initialTopicsToBeDeleted)
  // 用于其他线程与deleteTopicsThread删除线程同步
  val deleteTopicsCond = deleteLock.newCondition()
  // 用于标识Topic删除操作是否开始
  val deleteTopicStateChanged: AtomicBoolean = new AtomicBoolean(false)
  // 用于删除Topic的后台线程
  var deleteTopicsThread: DeleteTopicsThread = null

  // 从配置中读取是否开启了Topic删除功能
  val isDeleteTopicEnabled = controller.config.deleteTopicEnable // delete.topic.enable
  ...
}

// 用于删除主题的线程
class DeleteTopicsThread() extends ShutdownableThread(name = "delete-topics-thread-" + controller.config.brokerId, isInterruptible = false) {
  val zkUtils = controllerContext.zkUtils

  // 主要方法
  override def doWork() {
    // 等待删除操作开始
    awaitTopicDeletionNotification()

    if (!isRunning.get) // 判断状态
      return

    inLock(controllerContext.controllerLock) {

      // 收集将被删除的主题集合
      val topicsQueuedForDeletion = Set.empty[String] ++ topicsToBeDeleted

      if(!topicsQueuedForDeletion.isEmpty) // 将被删除的主题集合不为空
        info("Handling deletion for topics " + topicsQueuedForDeletion.mkString(","))

      topicsQueuedForDeletion.foreach { topic => // 遍历将要被删除的主题
      // if all replicas are marked as deleted successfully, then topic deletion is done
        if(controller.replicaStateMachine.areAllReplicasForTopicDeleted(topic)) { // 通过状态机判断主题的副本是否都删除了
          // clear up all state for this topic from controller cache and zookeeper
          // 完成主题的删除，会处理状态切换、Zookeeper相关存储及本地缓存的删除
          completeDeleteTopic(topic)
          info("Deletion of topic %s successfully completed".format(topic))
        } else {
          if(controller.replicaStateMachine.isAtLeastOneReplicaInDeletionStartedState(topic)) {
            // 检查该主题是否存在状态为ReplicaDeletionStarted的副本，如果存在则继续等待
            // ignore since topic deletion is in progress
            // 日志打印
            val replicasInDeletionStartedState = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionStarted)
            val replicaIds = replicasInDeletionStartedState.map(_.replica)
            val partitions = replicasInDeletionStartedState.map(r => TopicAndPartition(r.topic, r.partition))
            info("Deletion for replicas %s for partition %s of topic %s in progress".format(replicaIds.mkString(","),
              partitions.mkString(","), topic))
          } else {
            // if you come here, then no replica is in TopicDeletionStarted and all replicas are not in
            // TopicDeletionSuccessful. That means, that either given topic haven't initiated deletion
            // or there is at least one failed replica (which means topic deletion should be retried).
            // 任一副本处于ReplicaDeletionIneligible状态，则重置为OfflineReplica后重试
            if(controller.replicaStateMachine.isAnyReplicaInState(topic, ReplicaDeletionIneligible)) {
              // mark topic for deletion retry
              markTopicForDeletionRetry(topic)
            }
          }
        }
        // Try delete topic if it is eligible for deletion.
        // 检测当前Topic是否可以删除
        if(isTopicEligibleForDeletion(topic)) {
          info("Deletion of topic %s (re)started".format(topic))
          // topic deletion will be kicked off
          onTopicDeletion(Set(topic)) // 开始Topic的删除操作
        } else if(isTopicIneligibleForDeletion(topic)) {
          info("Not retrying deletion of topic %s at this time since it is marked ineligible for deletion".format(topic))
        }
      }
    }
  }
}

/**
 * This callback is invoked by the DeleteTopics thread with the list of topics to be deleted
 * It invokes the delete partition callback for all partitions of a topic.
 * The updateMetadataRequest is also going to set the leader for the topics being deleted to
 * {@link LeaderAndIsr#LeaderDuringDelete}. This lets each broker know that this topic is being deleted and can be
 * removed from their caches.
 */
private def onTopicDeletion(topics: Set[String]) {
  info("Topic deletion callback for %s".format(topics.mkString(",")))
  // send update metadata so that brokers stop serving data for topics to be deleted
  // 获取指定主题的所有分区
  val partitions = topics.flatMap(controllerContext.partitionsForTopic)
  // 向可用Broker发送UpdateMetadataRequest请求
  controller.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, partitions)
  // 按照Topic进行分组
  val partitionReplicaAssignmentByTopic = controllerContext.partitionReplicaAssignment.groupBy(p => p._1.topic)
  // 开始分区的删除操作
  topics.foreach { topic =>
    // 删除分区方法
    onPartitionDeletion(partitionReplicaAssignmentByTopic(topic).map(_._1).toSet)
  }
}

private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: AbstractRequestResponse, replicaId: Int) {
  val stopReplicaResponse = stopReplicaResponseObj.asInstanceOf[StopReplicaResponse]
  debug("Delete topic callback invoked for %s".format(stopReplicaResponse))
  // 响应集合
  val responseMap = stopReplicaResponse.responses.asScala
  // 整理响应中携带的每个分区的错误
  val partitionsInError =
    if (stopReplicaResponse.errorCode != Errors.NONE.code) responseMap.keySet
    else responseMap.filter { case (_, error) => error != Errors.NONE.code }.map(_._1).toSet
  // 整理得到每个副本的错误
  val replicasInError = partitionsInError.map(p => PartitionAndReplica(p.topic, p.partition, replicaId))
  inLock(controllerContext.controllerLock) {
    // move all the failed replicas to ReplicaDeletionIneligible
    // 使用failReplicaDeletion()方法处理响应中有错误信息的副本
    failReplicaDeletion(replicasInError)
    // 使用completeReplicaDeletion()方法处理响应中正常的副本
    if (replicasInError.size != responseMap.size) {
      // some replicas could have been successfully deleted
      val deletedReplicas = responseMap.keySet -- partitionsInError
      completeReplicaDeletion(deletedReplicas.map(p => PartitionAndReplica(p.topic, p.partition, replicaId)))
    }
  }
}

/**
 * Invoked when a broker that hosts replicas for topics to be deleted goes down. Also invoked when the callback for
 * StopReplicaResponse receives an error code for the replicas of a topic to be deleted. As part of this, the replicas
 * are moved from ReplicaDeletionStarted to ReplicaDeletionIneligible state. Also, the topic is added to the list of topics
 * ineligible for deletion until further notice. The delete topic thread is notified so it can retry topic deletion
 * if it has received a response for all replicas of a topic to be deleted
 * @param replicas Replicas for which deletion has failed
 */
def failReplicaDeletion(replicas: Set[PartitionAndReplica]) {
  if(isDeleteTopicEnabled) { // 检查是否开启了主题删除
    // 过滤出正在等待删除的主题的副本
    val replicasThatFailedToDelete = replicas.filter(r => isTopicQueuedUpForDeletion(r.topic))
    if(replicasThatFailedToDelete.size > 0) {
      val topics = replicasThatFailedToDelete.map(_.topic)
      debug("Deletion failed for replicas %s. Halting deletion for topics %s"
        .format(replicasThatFailedToDelete.mkString(","), topics))
      // 将这些副本的状态转换为ReplicaDeletionIneligible
      controller.replicaStateMachine.handleStateChanges(replicasThatFailedToDelete, ReplicaDeletionIneligible)
      // 标记这些副本所属的Topic不可用
      markTopicIneligibleForDeletion(topics)
      // 唤醒DeleteTopicsThread线程
      resumeTopicDeletionThread()
    }
  }
}

/**
 * Halt delete topic if -
 * 1. replicas being down
 * 2. partition reassignment in progress for some partitions of the topic
 * 3. preferred replica election in progress for some partitions of the topic
 * @param topics Topics that should be marked ineligible for deletion. No op if the topic is was not previously queued up for deletion
 */
def markTopicIneligibleForDeletion(topics: Set[String]) {
  if(isDeleteTopicEnabled) { // 判断是否开启了主题删除功能
    val newTopicsToHaltDeletion = topicsToBeDeleted & topics
    // 将传入的主题添加到topicsIneligibleForDeletion集合中，即表示他们不可被删除
    topicsIneligibleForDeletion ++= newTopicsToHaltDeletion
    if(newTopicsToHaltDeletion.size > 0)
      info("Halted deletion of topics %s".format(newTopicsToHaltDeletion.mkString(",")))
  }
}

/**
 * Invoked by the StopReplicaResponse callback when it receives no error code for a replica of a topic to be deleted.
 * As part of this, the replicas are moved from ReplicaDeletionStarted to ReplicaDeletionSuccessful state. The delete
 * topic thread is notified so it can tear down the topic if all replicas of a topic have been successfully deleted
 * @param replicas Replicas that were successfully deleted by the broker
 */
private def completeReplicaDeletion(replicas: Set[PartitionAndReplica]) {
  // 过滤得到成功删除的副本
  val successfullyDeletedReplicas = replicas.filter(r => isTopicQueuedUpForDeletion(r.topic))
  debug("Deletion successfully completed for replicas %s".format(successfullyDeletedReplicas.mkString(",")))
  // 将成功删除的副本的状态转换为ReplicaDeletionSuccessful
  controller.replicaStateMachine.handleStateChanges(successfullyDeletedReplicas, ReplicaDeletionSuccessful)
  // 唤醒DeleteTopicsThread线程
  resumeTopicDeletionThread()
}

private def completeDeleteTopic(topic: String) {
  // deregister partition change listener on the deleted topic. This is to prevent the partition change listener
  // firing before the new topic listener when a deleted topic gets auto created
  // 移除即将被删除的主题上的PartitionModificationsListener监听器
  partitionStateMachine.deregisterPartitionChangeListener(topic)
  // 从副本状态机中获取所有状态为ReplicaDeletionSuccessful的副本集
  val replicasForDeletedTopic = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)
  // controller will remove this replica from the state machine as well as its partition assignment cache
  // 使用副本状态机将这些状态为ReplicaDeletionSuccessful的副本集的状态转换为NonExistentReplica
  replicaStateMachine.handleStateChanges(replicasForDeletedTopic, NonExistentReplica)
  // 获取即将被删除主题的所有分区
  val partitionsForDeletedTopic = controllerContext.partitionsForTopic(topic)
  // move respective partition to OfflinePartition and NonExistentPartition state
  // 将即将被删除主题的所有分区的状态转换为OfflinePartition
  partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, OfflinePartition)
  // 将即将被删除主题的所有分区的状态转换为NonExistentPartition
  partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, NonExistentPartition)
  // 从topicsToBeDeleted中移除已经处理的主题
  topicsToBeDeleted -= topic
  // 从partitionsToBeDeleted中移除已经处理的主题的分区
  partitionsToBeDeleted.retain(_.topic != topic)
  // 维护Zookeeper中的信息
  val zkUtils = controllerContext.zkUtils
  zkUtils.zkClient.deleteRecursive(getTopicPath(topic))
  zkUtils.zkClient.deleteRecursive(getEntityConfigPath(ConfigType.Topic, topic))
  zkUtils.zkClient.delete(getDeleteTopicPath(topic))
  controllerContext.removeTopic(topic) // 情况ControllerContext中的缓存
}

/**
 * Called when leader intimates of isr change
  * IsrChangeNotificationListener用于监听/isr_change_notification路径下的子节点变化，
  * 当某些分区的ISR集合变化时通知整个集群中的所有Broker
 * @param controller
 */
class IsrChangeNotificationListener(controller: KafkaController) extends IZkChildListener with Logging {

  override def handleChildChange(parentPath: String, currentChildren: util.List[String]): Unit = {
    import scala.collection.JavaConverters._

    inLock(controller.controllerContext.controllerLock) {
      debug("[IsrChangeNotificationListener] Fired!!!")
      val childrenAsScala: mutable.Buffer[String] = currentChildren.asScala
      try {
        // 获取发生变化的主题分区
        val topicAndPartitions: immutable.Set[TopicAndPartition] = childrenAsScala.map(x => getTopicAndPartition(x)).flatten.toSet
        if (topicAndPartitions.nonEmpty) {
          // 从Zookeeper中读取指定分区的Leader副本、ISR集合等信息，更新ControllerContext
          controller.updateLeaderAndIsrCache(topicAndPartitions)
          // 向可用Broker发送UpdateMetadataRequest，更新它们的MetadataCache
          processUpdateNotifications(topicAndPartitions)
        }
      } finally {
        // delete processed children
        // 删除/isr_change_notification/partitions下已经处理的信息
        childrenAsScala.map(x => controller.controllerContext.zkUtils.deletePath(
          ZkUtils.IsrChangeNotificationPath + "/" + x))
      }
    }
  }

  private def processUpdateNotifications(topicAndPartitions: immutable.Set[TopicAndPartition]) {
    val liveBrokers: Seq[Int] = controller.controllerContext.liveOrShuttingDownBrokerIds.toSeq
    debug("Sending MetadataRequest to Brokers:" + liveBrokers + " for TopicAndPartitions:" + topicAndPartitions)
    // 向可用Broker发送UpdateMetadataRequest，更新它们的MetadataCache
    controller.sendUpdateMetadataRequest(liveBrokers, topicAndPartitions)
  }

  private def getTopicAndPartition(child: String): Set[TopicAndPartition] = {
    val changeZnode: String = ZkUtils.IsrChangeNotificationPath + "/" + child
    val (jsonOpt, stat) = controller.controllerContext.zkUtils.readDataMaybeNull(changeZnode)
    if (jsonOpt.isDefined) {
      val json = Json.parseFull(jsonOpt.get)

      json match {
        case Some(m) =>
          val topicAndPartitions: mutable.Set[TopicAndPartition] = new mutable.HashSet[TopicAndPartition]()
          val isrChanges = m.asInstanceOf[Map[String, Any]]
          val topicAndPartitionList = isrChanges("partitions").asInstanceOf[List[Any]]
          topicAndPartitionList.foreach {
            case tp =>
              val topicAndPartition = tp.asInstanceOf[Map[String, Any]]
              val topic = topicAndPartition("topic").asInstanceOf[String]
              val partition = topicAndPartition("partition").asInstanceOf[Int]
              topicAndPartitions += TopicAndPartition(topic, partition)
          }
          topicAndPartitions
        case None =>
          error("Invalid topic and partition JSON: " + jsonOpt.get + " in ZK: " + changeZnode)
          Set.empty
      }
    } else {
      Set.empty
    }
  }
}

/**
  * 监听/brokers/topics/[topic_name]节点中的数据变化，主要用于监听一个Topic的分区变化。
  * 并不对分区的删除进行处理，Topic的分区数量是不能减少的。
  */
class PartitionModificationsListener(topic: String) extends IZkDataListener with Logging {
  this.logIdent = "[AddPartitionsListener on " + controller.config.brokerId + "]: "

  @throws(classOf[Exception])
  def handleDataChange(dataPath : String, data: Object) {
    inLock(controllerContext.controllerLock) {
      try {
        info(s"Partition modification triggered $data for path $dataPath")
        // 从Zookeeper中获取Topic的Partition记录
        val partitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(List(topic))
        // 过滤出新增分区的记录
        val partitionsToBeAdded = partitionReplicaAssignment.filter(p =>
          !controllerContext.partitionReplicaAssignment.contains(p._1))
        // 主题正在进行删除操作，输出日志后直接返回
        if(controller.deleteTopicManager.isTopicQueuedUpForDeletion(topic))
          error("Skipping adding partitions %s for topic %s since it is currently being deleted"
                .format(partitionsToBeAdded.map(_._1.partition).mkString(","), topic))
        else {
          if (partitionsToBeAdded.size > 0) {
            info("New partitions to be added %s".format(partitionsToBeAdded))
            // 将新增分区信息添加到ControllerContext中
            controllerContext.partitionReplicaAssignment.++=(partitionsToBeAdded)
            // 切换新增分区及其副本的状态，最终使其上线对外服务
            controller.onNewPartitionCreation(partitionsToBeAdded.keySet.toSet)
          }
        }
      } catch {
        case e: Throwable => error("Error while handling add partitions for data path " + dataPath, e )
      }
    }
  }

  @throws(classOf[Exception])
  def handleDataDeleted(parentPath : String) {
    // this is not implemented for partition change
  }
}

/**
 * Starts the preferred replica leader election for the list of partitions specified under
  * 负责监听ZooKeeper节点/admin/preferred_replica_election
  * 当通过PreferredReplicaLeaderElectionCommand命令指定某些分区需要进行“优先副本”选举时会将指定分区的信息写入该节点，
  * 从而触发PreferredReplicaElectionListener进行处理
  * 进行“优先副本”选举的目的是让分区的“优先副本”重新成为Leader副本，这是为了让Leader副本在整个集群中分布得更加均衡
 * /admin/preferred_replica_election -
 */
class PreferredReplicaElectionListener(controller: KafkaController) extends IZkDataListener with Logging {
  this.logIdent = "[PreferredReplicaElectionListener on " + controller.config.brokerId + "]: "
  val zkUtils = controller.controllerContext.zkUtils
  val controllerContext = controller.controllerContext

  /**
   * Invoked when some partitions are reassigned by the admin command
   * @throws Exception On any error.
   */
  @throws(classOf[Exception])
  def handleDataChange(dataPath: String, data: Object) {
    debug("Preferred replica election listener fired for path %s. Record partitions to undergo preferred replica election %s"
            .format(dataPath, data.toString))
    inLock(controllerContext.controllerLock) {
      // 获取需要进行"优先副本"选举的TopicAndPartition列表
      val partitionsForPreferredReplicaElection = PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(data.toString)
      if(controllerContext.partitionsUndergoingPreferredReplicaElection.size > 0)
        info("These partitions are already undergoing preferred replica election: %s"
          .format(controllerContext.partitionsUndergoingPreferredReplicaElection.mkString(",")))
      // 过滤掉正在进行"优先副本"选举的分区
      val partitions = partitionsForPreferredReplicaElection -- controllerContext.partitionsUndergoingPreferredReplicaElection
      // 过滤掉待删除Topic的分区
      val partitionsForTopicsToBeDeleted = partitions.filter(p => controller.deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
      if(partitionsForTopicsToBeDeleted.size > 0) {
        error("Skipping preferred replica election for partitions %s since the respective topics are being deleted"
          .format(partitionsForTopicsToBeDeleted))
      }
      // 对剩余的分区调用onPreferredReplicaElection()方法进行"优先副本"的选举
      controller.onPreferredReplicaElection(partitions -- partitionsForTopicsToBeDeleted)
    }
  }

  /**
   * @throws Exception
   *             On any error.
   */
  @throws(classOf[Exception])
  def handleDataDeleted(dataPath: String) {
  }
}

/**
 * Starts the partition reassignment process unless -
 * 1. Partition previously existed
 * 2. New replicas are the same as existing replicas
 * 3. Any replica in the new set of replicas are dead
 * If any of the above conditions are satisfied, it logs an error and removes the partition from list of reassigned
 * partitions.
  *
  * PartitionsReassignedListener监听的ZooKeeper节点是/admin/reassign_partitions。
  * 当管理人员通过ReassignPartitionsCommand命令指定某些分区需要重新分配副本时，
  * 会将指定分区的信息写入该节点，从而触发PartitionsReassignedListener进行处理。
 */
class PartitionsReassignedListener(controller: KafkaController) extends IZkDataListener with Logging {
  this.logIdent = "[PartitionsReassignedListener on " + controller.config.brokerId + "]: "
  val zkUtils = controller.controllerContext.zkUtils
  val controllerContext = controller.controllerContext

  /**
   * Invoked when some partitions are reassigned by the admin command
   * @throws Exception On any error.
   */
  @throws(classOf[Exception])
  def handleDataChange(dataPath: String, data: Object) {
    debug("Partitions reassigned listener fired for path %s. Record partitions to be reassigned %s"
      .format(dataPath, data))
    // 从Zookeeper的/admin/reassign_partitions节点下读取分区的副本重分配信息
    val partitionsReassignmentData = zkUtils.parsePartitionReassignmentData(data.toString)
    // 过滤正在进行重分配的分区
    val partitionsToBeReassigned = inLock(controllerContext.controllerLock) {
      partitionsReassignmentData.filterNot(p => controllerContext.partitionsBeingReassigned.contains(p._1))
    }
    partitionsToBeReassigned.foreach { partitionToBeReassigned =>
      inLock(controllerContext.controllerLock) {
        // 检测分区所属的Topic是否为待删除Topic
        if(controller.deleteTopicManager.isTopicQueuedUpForDeletion(partitionToBeReassigned._1.topic)) {
          error("Skipping reassignment of partition %s for topic %s since it is currently being deleted"
            .format(partitionToBeReassigned._1, partitionToBeReassigned._1.topic))
          controller.removePartitionFromReassignedPartitions(partitionToBeReassigned._1)
        } else {
          val context = new ReassignedPartitionsContext(partitionToBeReassigned._2)
          // 为副本重新分配做一些准备工作
          controller.initiateReassignReplicasForTopicPartition(partitionToBeReassigned._1, context)
        }
      }
    }
  }

  /**
   * Called when the leader information stored in zookeeper has been delete. Try to elect as the leader
   * @throws Exception
   *             On any error.
   */
  @throws(classOf[Exception])
  def handleDataDeleted(dataPath: String) {
  }
}

// 对需要进行副本重新分配的分区做准备工作
def initiateReassignReplicasForTopicPartition(topicAndPartition: TopicAndPartition,
                                      reassignedPartitionContext: ReassignedPartitionsContext) {
  ...
  // 为分区注册ReassignedPartitionsIsrChangeListener
  watchIsrChangesForReassignedPartition(topic, partition, reassignedPartitionContext)
  ...
}

// 为指定的需要进行副本重分配的分区绑定ReassignedPartitionsIsrChangeListener监听器
private def watchIsrChangesForReassignedPartition(topic: String,
                                                  partition: Int,
                                                  reassignedPartitionContext: ReassignedPartitionsContext) {
  // 获取分配给指定分区的新的AR集合
  val reassignedReplicas = reassignedPartitionContext.newReplicas
  // 根据分区所属主题、分区ID以及新的AR集合创建ReassignedPartitionsIsrChangeListener监听器
  val isrChangeListener = new ReassignedPartitionsIsrChangeListener(this, topic, partition,
    reassignedReplicas.toSet)
  // 将监听器绑定到分区对应的ReassignedPartitionsContext对象上
  reassignedPartitionContext.isrChangeListener = isrChangeListener
  // register listener on the leader and isr path to wait until they catch up with the current leader
  // 在Zookeeper的/brokers/topics/[topic_name]/partitions/[partition_id]/state路径上注册刚刚创建的监听器
  zkUtils.zkClient.subscribeDataChanges(getTopicPartitionLeaderAndIsrPath(topic, partition), isrChangeListener)
}

/**
  * 每个分区都会注册一个对应的ReassignedPartitionsIsrChangeListener监听器
  * 监听路径为/broker/topics/[topic_name]/partitions/[partitionId]/state
  * @param controller KafkaController
  * @param topic 主题
  * @param partition 分区
  * @param reassignedReplicas 新分配的AR集合
  */
class ReassignedPartitionsIsrChangeListener(controller: KafkaController, topic: String, partition: Int,
                                            reassignedReplicas: Set[Int])
  extends IZkDataListener with Logging {
  this.logIdent = "[ReassignedPartitionsIsrChangeListener on controller " + controller.config.brokerId + "]: "
  val zkUtils = controller.controllerContext.zkUtils
  val controllerContext = controller.controllerContext

  /**
   * Invoked when some partitions need to move leader to preferred replica
   * @throws Exception On any error.
   */
  @throws(classOf[Exception])
  def handleDataChange(dataPath: String, data: Object) {
    inLock(controllerContext.controllerLock) {
      debug("Reassigned partitions isr change listener fired for path %s with children %s".format(dataPath, data))
      val topicAndPartition = TopicAndPartition(topic, partition)
      try {
        // check if this partition is still being reassigned or not
        // 判断当前监听器所绑定的分区是否正在进行副本重新分配
        controllerContext.partitionsBeingReassigned.get(topicAndPartition) match {
          case Some(reassignedPartitionContext) => // 正在进行
            // need to re-read leader and isr from zookeeper since the zkclient callback doesn't return the Stat object
            // 从Zookeeper中获取当前分区的Leader副本、ISR集合等信息
            val newLeaderAndIsrOpt = zkUtils.getLeaderAndIsrForPartition(topic, partition)
            newLeaderAndIsrOpt match { // Zookeeper中能获取到Leader相关信息
              case Some(leaderAndIsr) => // check if new replicas have joined ISR
                // 检查RAR集合中的所有副本是否已经加入了ISR集合
                // 取RAR与ISR的并集
                val caughtUpReplicas = reassignedReplicas & leaderAndIsr.isr.toSet
                // 检查RAR与上述并集是否相同，如果相同说明RAR的副本都进入了ISR集合
                if(caughtUpReplicas == reassignedReplicas) {
                  // resume the partition reassignment process
                  info("%d/%d replicas have caught up with the leader for partition %s being reassigned."
                    .format(caughtUpReplicas.size, reassignedReplicas.size, topicAndPartition) +
                    "Resuming partition reassignment")
                  // RAR集合的副本都进入了ISR集合，进行副本重分配操作
                  controller.onPartitionReassignment(topicAndPartition, reassignedPartitionContext)
                }
                else {
                  info("%d/%d replicas have caught up with the leader for partition %s being reassigned."
                    .format(caughtUpReplicas.size, reassignedReplicas.size, topicAndPartition) +
                    "Replica(s) %s still need to catch up".format((reassignedReplicas -- leaderAndIsr.isr.toSet).mkString(",")))
                }
              case None => error("Error handling reassignment of partition %s to replicas %s as it was never created"
                .format(topicAndPartition, reassignedReplicas.mkString(",")))
            }
          case None => // 没有进行，结束操作
        }
      } catch {
        case e: Throwable => error("Error while handling partition reassignment", e)
      }
    }
  }

  /**
   * @throws Exception
   *             On any error.
   */
  @throws(classOf[Exception])
  def handleDataDeleted(dataPath: String) {
  }
}