#!/bin/bash

if [ $# -lt 1 ]; # 检查参数个数
then # 打印使用方法
    echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
    exit 1
fi

# 获取当前脚本所在的路径
base_dir=$(dirname $0)

# 设置Log4j相关的环境变量
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

# 设置JVM内存
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi

EXTRA_ARGS="-name kafkaServer -loggc"

# 检测第一个参数是否为"-daemon"
COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    # 左移参数列表，即删除"-daemon"参数
    shift
    ;;
  *)
    ;;
esac

# 调用kafka-run-class.sh脚本，传入参数以及主类
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

#!/bin/bash
# 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.

# 检测参数是否合法
if [ $# -lt 1 ];
then
  # 打印使用方式
  echo "USAGE: $0 [-daemon] [-name servicename] [-loggc] classname [opts]"
  exit 1
fi

# 检测INCLUDE_TEST_JARS变量是否为空，如果为空将其设置为false
if [ -z "$INCLUDE_TEST_JARS" ]; then
  INCLUDE_TEST_JARS=false
fi

# Exclude jars not necessary for running commands.
# 定义should_include_file函数，用于检测CLASSPATH是否需要包含指定的文件
regex="(-(test|src|scaladoc|javadoc)\.jar|jar.asc)$"
should_include_file() {
  if [ "$INCLUDE_TEST_JARS" = true ]; then
    return 0
  fi
  file=$1
  if [ -z "$(echo "$file" | egrep "$regex")" ] ; then
    return 0
  else
    return 1
  fi
}

# 获取脚本所在目录的上层目录
base_dir=$(dirname $0)/..

# 检测并设置SCALA_VERSION
if [ -z "$SCALA_VERSION" ]; then
    SCALA_VERSION=2.10.6
fi

# 检测并设置SCALA_BINARY_VERSION
if [ -z "$SCALA_BINARY_VERSION" ]; then
    SCALA_BINARY_VERSION=2.10
fi

# run ./gradlew copyDependantLibs to get all dependant jars in a local dir
# 检测$base_dir下的多个目录，根据should_include_file函数设置CLASSPATH
shopt -s nullglob
for dir in "$base_dir"/core/build/dependant-libs-${SCALA_VERSION}*;
do
  if [ -z "$CLASSPATH" ] ; then
    CLASSPATH="$dir/*"
  else
    CLASSPATH="$CLASSPATH:$dir/*"
  fi
done

for file in "$base_dir"/examples/build/libs/kafka-examples*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for file in "$base_dir"/clients/build/libs/kafka-clients*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for file in "$base_dir"/streams/build/libs/kafka-streams*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for file in "$base_dir"/streams/build/dependant-libs-${SCALA_VERSION}/rocksdb*.jar;
do
  CLASSPATH="$CLASSPATH":"$file"
done

for file in "$base_dir"/tools/build/libs/kafka-tools*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for dir in "$base_dir"/tools/build/dependant-libs-${SCALA_VERSION}*;
do
  CLASSPATH="$CLASSPATH:$dir/*"
done

for cc_pkg in "api" "runtime" "file" "json" "tools"
do
  for file in "$base_dir"/connect/${cc_pkg}/build/libs/connect-${cc_pkg}*.jar;
  do
    if should_include_file "$file"; then
      CLASSPATH="$CLASSPATH":"$file"
    fi
  done
  if [ -d "$base_dir/connect/${cc_pkg}/build/dependant-libs" ] ; then
    CLASSPATH="$CLASSPATH:$base_dir/connect/${cc_pkg}/build/dependant-libs/*"
  fi
done

# classpath addition for release
for file in "$base_dir"/libs/*;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done

for file in "$base_dir"/core/build/libs/kafka_${SCALA_BINARY_VERSION}*.jar;
do
  if should_include_file "$file"; then
    CLASSPATH="$CLASSPATH":"$file"
  fi
done
shopt -u nullglob

# JMX settings
# JMX的相关设置
if [ -z "$KAFKA_JMX_OPTS" ]; then
  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
fi

# JMX port to use
# 设置JMX的端口
if [  $JMX_PORT ]; then
  KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi

# Log directory to use
# 指定存放日志文件和索引文件的目录，默认是$KAFKA_HOME/logs
if [ "x$LOG_DIR" = "x" ]; then
    LOG_DIR="$base_dir/logs"
fi

# Log4j settings
# Log4j的相关设置
if [ -z "$KAFKA_LOG4J_OPTS" ]; then
  # Log to console. This is a tool.
  KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/config/tools-log4j.properties"
else
  # create logs directory
  if [ ! -d "$LOG_DIR" ]; then
    mkdir -p "$LOG_DIR"
  fi
fi

KAFKA_LOG4J_OPTS="-Dkafka.logs.dir=$LOG_DIR $KAFKA_LOG4J_OPTS"

# Generic jvm settings you want to add
if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS=""
fi

# Set Debug options if enabled
# 检测是否以Debug模式启动
if [ "x$KAFKA_DEBUG" != "x" ]; then

    # Use default ports
    DEFAULT_JAVA_DEBUG_PORT="5005"

    if [ -z "$JAVA_DEBUG_PORT" ]; then
        JAVA_DEBUG_PORT="$DEFAULT_JAVA_DEBUG_PORT"
    fi

    # Use the defaults if JAVA_DEBUG_OPTS was not set
    DEFAULT_JAVA_DEBUG_OPTS="-agentlib:jdwp=transport=dt_socket,server=y,suspend=${DEBUG_SUSPEND_FLAG:-n},address=$JAVA_DEBUG_PORT"
    if [ -z "$JAVA_DEBUG_OPTS" ]; then
        JAVA_DEBUG_OPTS="$DEFAULT_JAVA_DEBUG_OPTS"
    fi

    echo "Enabling Java debug options: $JAVA_DEBUG_OPTS"
    KAFKA_OPTS="$JAVA_DEBUG_OPTS $KAFKA_OPTS"
fi

# Which java to use
# 检测JAVA_HOME环境变量
if [ -z "$JAVA_HOME" ]; then
  JAVA="java"
else
  JAVA="$JAVA_HOME/bin/java"
fi

# Memory options
# 配置JVM内存
if [ -z "$KAFKA_HEAP_OPTS" ]; then
  KAFKA_HEAP_OPTS="-Xmx256M"
fi

# JVM performance options
# 对JVM进行一些优化配置，默认使用G1垃圾回收器
if [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; then
  KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC -Djava.awt.headless=true"
fi

# 处理参数name、loggc、daemon三个参数
while [ $# -gt 0 ]; do
  COMMAND=$1
  case $COMMAND in
    -name)
      DAEMON_NAME=$2
      CONSOLE_OUTPUT_FILE=$LOG_DIR/$DAEMON_NAME.out
      shift 2
      ;;
    -loggc)
      if [ -z "$KAFKA_GC_LOG_OPTS" ]; then
        GC_LOG_ENABLED="true"
      fi
      shift
      ;;
    -daemon)
      DAEMON_MODE="true"
      shift
      ;;
    *)
      break
      ;;
  esac
done

# GC options
# 调整JVM GC相关的参数
GC_FILE_SUFFIX='-gc.log'
GC_LOG_FILE_NAME=''
if [ "x$GC_LOG_ENABLED" = "xtrue" ]; then
  GC_LOG_FILE_NAME=$DAEMON_NAME$GC_FILE_SUFFIX
  KAFKA_GC_LOG_OPTS="-Xloggc:$LOG_DIR/$GC_LOG_FILE_NAME -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps "
fi

# Launch mode
# 根据$DAEMON_MODE的值，决定是否后台启动
# 这里就是主要的执行业务代码了，会使用java命令执行传入的参数中携带的主类的main方法
if [ "x$DAEMON_MODE" = "xtrue" ]; then
  nohup $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
  exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@"
fi

def main(args: Array[String]): Unit = {

  // 解析参数，TopicCommandOptions支持list、describe、create、alter和delete五种操作
  val opts = new TopicCommandOptions(args)

  // 检查参数长度
  if(args.length == 0)
    CommandLineUtils.printUsageAndDie(opts.parser, "Create, delete, describe, or change a topic.")

  // should have exactly one action
  // 获取对应的操作
  val actions = Seq(opts.createOpt, opts.listOpt, opts.alterOpt, opts.describeOpt, opts.deleteOpt).count(opts.options.has _)
  if(actions != 1)
    CommandLineUtils.printUsageAndDie(opts.parser, "Command must include exactly one action: --list, --describe, --create, --alter or --delete")

  // 检查参数
  opts.checkArgs()

  // 获取Zookeeper连接
  val zkUtils = ZkUtils(opts.options.valueOf(opts.zkConnectOpt),
                        30000,
                        30000,
                        JaasUtils.isZkSecurityEnabled())
  var exitCode = 0
  try {
    if(opts.options.has(opts.createOpt)) // --create参数，创建Topic
      createTopic(zkUtils, opts)
    else if(opts.options.has(opts.alterOpt)) // --alter参数，修改Topic
      alterTopic(zkUtils, opts)
    else if(opts.options.has(opts.listOpt)) // --list参数，列出Topic
      listTopics(zkUtils, opts)
    else if(opts.options.has(opts.describeOpt)) // --describe参数，查询Topic详细信息
      describeTopic(zkUtils, opts)
    else if(opts.options.has(opts.deleteOpt)) // --delete参数，删除Topic
      deleteTopic(zkUtils, opts)
  } catch {
    case e: Throwable =>
      println("Error while executing topic command : " + e.getMessage)
      error(Utils.stackTrace(e))
      exitCode = 1
  } finally {
    zkUtils.close()
    System.exit(exitCode)
  }

}

def createTopic(zkUtils: ZkUtils, opts: TopicCommandOptions) {
  // 获取--topic参数
  val topic = opts.options.valueOf(opts.topicOpt)
  // 将--config参数解析成Properties对象
  val configs = parseTopicConfigsToBeAdded(opts)
  // 读取--if-not-exists参数
  val ifNotExists = if (opts.options.has(opts.ifNotExistsOpt)) true else false
  // 检测Topic名称中是否包含"."或"_"字符，若包含则输出警告信息
  if (Topic.hasCollisionChars(topic))
    println("WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.")
  try {
    // 检测是否有--replica-assignment参数
    if (opts.options.has(opts.replicaAssignmentOpt)) {
      /**
        * --replica-assignment参数的格式类似于：0:1:2,3:4:5,6:7:8，含义如下：
        * 1. 编号为0的分区分配在Broker-0、Broker-1和Broker-2上；
        * 2. 编号为1的分区分配在Broker-3、Broker-4和Broker-5上；
        * 3. 编号为2的分区分配在Broker-6、Broker-7和Broker-8上；
        *
        * 这里将--replica-assignment参数内容解析成Map[Int, Seq[Int]]格式，
        * 其key为分区的编号，value是其副本所分配的BrokerId
        */
      val assignment = parseReplicaAssignment(opts.options.valueOf(opts.replicaAssignmentOpt))
      // 检测max.message.bytes配置参数并给出提示
      warnOnMaxMessagesChange(configs, assignment.valuesIterator.next().length)
      // 对Topic名称和副本分配结果进行一系列的检测，并写入Zookeeper中
      AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, assignment, configs, update = false)
    } else {
      // 如果进行副本自动分配，必须指定--partitions和--replication-factor参数
      CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.partitionsOpt, opts.replicationFactorOpt)
      // 获取--partitions参数的值
      val partitions = opts.options.valueOf(opts.partitionsOpt).intValue
      // 获取--replication-factor参数的值
      val replicas = opts.options.valueOf(opts.replicationFactorOpt).intValue
      // 检测max.message.bytes配置参数并给出提示
      warnOnMaxMessagesChange(configs, replicas)
      // 根据--disable-rack-aware参数决定分配副本时是否考虑机架信息
      val rackAwareMode = if (opts.options.has(opts.disableRackAware)) RackAwareMode.Disabled
                          else RackAwareMode.Enforced

      // 自动分配副本，并写入Zookeeper
      AdminUtils.createTopic(zkUtils, topic, partitions, replicas, configs, rackAwareMode)
    }
    println("Created topic \"%s\".".format(topic))
  } catch  {
    case e: TopicExistsException => if (!ifNotExists) throw e
  }
}

// 创建并更新Zookeeper中的主题分区副本分配信息
def createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils: ZkUtils,
                                                   topic: String,
                                                   partitionReplicaAssignment: Map[Int, Seq[Int]],
                                                   config: Properties = new Properties,
                                                   update: Boolean = false) {
  // validate arguments
  // 检测Topic名称是否符合要求
  Topic.validate(topic)

  // --replica-assignment参数指定的每个分区的副本数应该相同
  require(partitionReplicaAssignment.values.map(_.size).toSet.size == 1, "All partitions should have the same number of replicas.")

  // 得到/brokers/topics/[topic_name]主题路径
  val topicPath = getTopicPath(topic)

  if (!update) {
    if (zkUtils.zkClient.exists(topicPath)) // Topic已存在
      throw new TopicExistsException("Topic \"%s\" already exists.".format(topic))
    else if (Topic.hasCollisionChars(topic)) { // Topic名称存在"."或"_"
      // 获取所有的Topic名称集合，打印提示出可能与新建Topic名称产生冲突的Topic
      val allTopics = zkUtils.getAllTopics()
      val collidingTopics = allTopics.filter(t => Topic.hasCollision(topic, t))
      if (collidingTopics.nonEmpty) {
        throw new InvalidTopicException("Topic \"%s\" collides with existing topics: %s".format(topic, collidingTopics.mkString(", ")))
      }
    }
  }

  // 日志打印
  partitionReplicaAssignment.values.foreach(reps => require(reps.size == reps.toSet.size, "Duplicate replica assignment found: "  + partitionReplicaAssignment))

  // Configs only matter if a topic is being created. Changing configs via AlterTopic is not supported
  if (!update) {
    // write out the config if there is any, this isn't transactional with the partition assignments
    // 验证配置
    LogConfig.validate(config)
    // 写入配置到Zookeeper
    writeEntityConfig(zkUtils, ConfigType.Topic, topic, config)
  }

  // create the partition assignment
  // 创建分区分配
  writeTopicPartitionAssignment(zkUtils, topic, partitionReplicaAssignment, update)
}

// 写入主题分区副本分配信息到Zookeeper
private def writeTopicPartitionAssignment(zkUtils: ZkUtils, topic: String, replicaAssignment: Map[Int, Seq[Int]], update: Boolean) {
  try {
    // 得到/brokers/topics/[topic_name]路径
    val zkPath = getTopicPath(topic)
    // 格式化副本分配信息为JSON字符串
    val jsonPartitionData = zkUtils.replicaAssignmentZkData(replicaAssignment.map(e => (e._1.toString -> e._2)))

    if (!update) {
      info("Topic creation " + jsonPartitionData.toString)
      // 创建持久节点
      zkUtils.createPersistentPath(zkPath, jsonPartitionData)
    } else {
      // 更新持久节点
      info("Topic update " + jsonPartitionData.toString)
      zkUtils.updatePersistentPath(zkPath, jsonPartitionData)
    }
    debug("Updated path %s with %s for replica assignment".format(zkPath, jsonPartitionData))
  } catch {
    case e: ZkNodeExistsException => throw new TopicExistsException("topic %s already exists".format(topic))
    case e2: Throwable => throw new AdminOperationException(e2.toString)
  }
}

private def assignReplicasToBrokersRackAware(nPartitions: Int,
                                             replicationFactor: Int,
                                             brokerMetadatas: Seq[BrokerMetadata],
                                             fixedStartIndex: Int,
                                             startPartitionId: Int): Map[Int, Seq[Int]] = {
  // 对机架信息进行转换，得到的字典中，键是Broker的ID，值为所在的机架名称
  val brokerRackMap = brokerMetadatas.collect { case BrokerMetadata(id, Some(rack)) =>
    id -> rack
  }.toMap
  // 统计机架个数
  val numRacks = brokerRackMap.values.toSet.size
  // 基于机架信息生产一个Broker列表，轮询每个机架上的Broker，不同机架上的Broker交替出现
  val arrangedBrokerList = getRackAlternatedBrokerList(brokerRackMap)
  // 所有机架上的Broker数量
  val numBrokers = arrangedBrokerList.size
  // 用于记录副本分配结果
  val ret = mutable.Map[Int, Seq[Int]]()
  // 选择起始Broker进行分配
  val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)
  // 选择起始分区
  var currentPartitionId = math.max(0, startPartitionId)
  // 指定副本的间隔
  var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)
  // 遍历所有分区
  for (_ <- 0 until nPartitions) {
    if (currentPartitionId > 0 && (currentPartitionId % arrangedBrokerList.size == 0))
      // 递增nextReplicaShift
      nextReplicaShift += 1
    // 计算"优先副本"所在的Broker
    val firstReplicaIndex = (currentPartitionId + startIndex) % arrangedBrokerList.size
    val leader = arrangedBrokerList(firstReplicaIndex)
    // 记录"优先副本"所在的Broker
    val replicaBuffer = mutable.ArrayBuffer(leader)
    // 记录以及分配了当前分区的副本的机架信息
    val racksWithReplicas = mutable.Set(brokerRackMap(leader))
    // 记录已经分配了当前分区的副本的Broker信息
    val brokersWithReplicas = mutable.Set(leader)
    var k = 0
    // 遍历分配剩余的副本
    for (_ <- 0 until replicationFactor - 1) {
      var done = false
      while (!done) {
        // 通过replicaIndex()方法计算当前副本所在的Broker
        val broker = arrangedBrokerList(replicaIndex(firstReplicaIndex, nextReplicaShift * numRacks, k, arrangedBrokerList.size))
        val rack = brokerRackMap(broker)
        // Skip this broker if
        // 1. there is already a broker in the same rack that has assigned a replica AND there is one or more racks
        //    that do not have any replica, or
        // 2. the broker has already assigned a replica AND there is one or more brokers that do not have replica assigned
        /**
          * 检测是否跳过此Broker，满足以下之一就跳过：
          * 1. 当前机架上已经分配过其他副本，而且存在机架还未 分配副本；
          * 2. 当前Broker上已经分配过其他副本，而且存在其他Broker还未分配副本。
          */
        if ((!racksWithReplicas.contains(rack) || racksWithReplicas.size == numRacks)
            && (!brokersWithReplicas.contains(broker) || brokersWithReplicas.size == numBrokers)) {
          // 记录分配结果
          replicaBuffer += broker
          // 记录此机架已经分配了当前分区的副本
          racksWithReplicas += rack
          // 记录此Broker已经分配了当前分区的副本
          brokersWithReplicas += broker
          // 标识此副本的分配完成
          done = true
        }
        k += 1
      }
    }
    ret.put(currentPartitionId, replicaBuffer)
    currentPartitionId += 1
  }
  ret
}

/**
  * Given broker and rack information, returns a list of brokers alternated by the rack. Assume
  * this is the rack and its brokers:
  *
  * rack1: 0, 1, 2
  * rack2: 3, 4, 5
  * rack3: 6, 7, 8
  *
  * This API would return the list of 0, 3, 6, 1, 4, 7, 2, 5, 8
  *
  * This is essential to make sure that the assignReplicasToBrokers API can use such list and
  * assign replicas to brokers in a simple round-robin fashion, while ensuring an even
  * distribution of leader and replica counts on each broker and that replicas are
  * distributed to all racks.
  */
private[admin] def getRackAlternatedBrokerList(brokerRackMap: Map[Int, String]): IndexedSeq[Int] = {
  // 得到 机架 -> 该机架上的Broker的ID集合的迭代器
  val brokersIteratorByRack = getInverseMap(brokerRackMap).map { case (rack, brokers) =>
    (rack, brokers.toIterator)
  }
  // 对机架进行排序
  val racks = brokersIteratorByRack.keys.toArray.sorted
  // 定义结果数组
  val result = new mutable.ArrayBuffer[Int]
  // 机架索引
  var rackIndex = 0
  /**
    * brokerRackMap的大小即为Broker的数量。
    * 这里的操作实现如下：
    * 因为之前排过序了，所以brokersIteratorByRack中机架对应的BrokerID是有序的；
    * 按照Broker的数量为遍历次数进行循环，轮流取每个机架上的Broker中的一个。
    * 例如，有以下Broker分配信息：
    *     rack1: 0, 1, 2
    *     rack2: 3, 4, 5
    *     rack3: 6, 7, 8
    * 一共9个Broker，那么会遍历9次：
    * 第1次：取rack1，取rack1上的Broker-0
    * 第2次：取rack2，取rack1上的Broker-3
    * 第3次：取rack3，取rack1上的Broker-6
    * 第4次：取rack1，取rack2上的Broker-1
    * 第5次：取rack2，取rack2上的Broker-5
    * 第6次：取rack3，取rack2上的Broker-7
    * 第7次：取rack1，取rack3上的Broker-2
    * 第8次：取rack2，取rack3上的Broker-5
    * 第9次：取rack3，取rack3上的Broker-8
    *
    * 最终得到的Broker ID序列为：
    * 0, 3, 6, 1, 4, 7, 2, 5, 8
    */
  while (result.size < brokerRackMap.size) {
    // 获取机架对应的BrokerID集合迭代器
    val rackIterator = brokersIteratorByRack(racks(rackIndex))
    // 取其中一个Broker
    if (rackIterator.hasNext)
      result += rackIterator.next()
    // 机架索引自增
    rackIndex = (rackIndex + 1) % racks.length
  }
  result
}

def alterTopic(zkUtils: ZkUtils, opts: TopicCommandOptions) {
  // 从Zookeeper中获取与--topic参数正则匹配的Topic集合
  val topics = getTopics(zkUtils, opts)
  // 读取--if-exists参数
  val ifExists = if (opts.options.has(opts.ifExistsOpt)) true else false
  if (topics.length == 0 && !ifExists) {
    throw new IllegalArgumentException("Topic %s does not exist on ZK path %s".format(opts.options.valueOf(opts.topicOpt),
        opts.options.valueOf(opts.zkConnectOpt)))
  }
  topics.foreach { topic =>
    // 修改Topic配置项信息的功能路径为/config/topics/[topic_name]
    val configs = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic, topic)
    // 判断是否包含--config配置或--delete-config配置
    if(opts.options.has(opts.configOpt) || opts.options.has(opts.deleteConfigOpt)) {
      println("WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.")
      println("         Going forward, please use kafka-configs.sh for this functionality")

      // 得到新添加的和将删除的配置并进行更新
      val configsToBeAdded = parseTopicConfigsToBeAdded(opts)
      val configsToBeDeleted = parseTopicConfigsToBeDeleted(opts)
      // compile the final set of configs
      configs.putAll(configsToBeAdded)
      configsToBeDeleted.foreach(config => configs.remove(config))
      // 修改Zookeeper中的主题配置信息
      AdminUtils.changeTopicConfig(zkUtils, topic, configs)
      println("Updated config for topic \"%s\".".format(topic))
    }

    // 副本重新分配的功能
    // 检测是否包含--partitions参数
    if(opts.options.has(opts.partitionsOpt)) {
      // 不可修改__consumer_offsets主题，会抛出异常
      if (topic == TopicConstants.GROUP_METADATA_TOPIC_NAME) {
        throw new IllegalArgumentException("The number of partitions for the offsets topic cannot be changed.")
      }
      println("WARNING: If partitions are increased for a topic that has a key, the partition " +
        "logic or ordering of the messages will be affected")
      // 获取--partitions参数
      val nPartitions = opts.options.valueOf(opts.partitionsOpt).intValue
      // 获取--replica-assignment参数
      val replicaAssignmentStr = opts.options.valueOf(opts.replicaAssignmentOpt)
      // 完成分区数量的增加以及副本分配
      AdminUtils.addPartitions(zkUtils, topic, nPartitions, replicaAssignmentStr)
      println("Adding partitions succeeded!")
    }
  }
}

def addPartitions(zkUtils: ZkUtils,
                  topic: String,
                  numPartitions: Int = 1,
                  replicaAssignmentStr: String = "",
                  checkBrokerAvailable: Boolean = true,
                  rackAwareMode: RackAwareMode = RackAwareMode.Enforced) {
  // 从Zookeeper获取此Topic当前的副本分配情况
  val existingPartitionsReplicaList = zkUtils.getReplicaAssignmentForTopics(List(topic))
  if (existingPartitionsReplicaList.size == 0)
    throw new AdminOperationException("The topic %s does not exist".format(topic))

  // 获取ID为0的分区的副本分配情况
  val existingReplicaListForPartitionZero = existingPartitionsReplicaList.find(p => p._1.partition == 0) match {
    case None => throw new AdminOperationException("the topic does not have partition with id 0, it should never happen")
    case Some(headPartitionReplica) => headPartitionReplica._2
  }

  // 获取ID未0的分区的副本数量
  val partitionsToAdd = numPartitions - existingPartitionsReplicaList.size
  if (partitionsToAdd <= 0)
    throw new AdminOperationException("The number of partitions for a topic can only be increased")

  // create the new partition replication list
  // 只能增加分区数量，如果指定的分区数量小于当前分区数，则抛出异常
  val brokerMetadatas = getBrokerMetadatas(zkUtils, rackAwareMode)
  val newPartitionReplicaList =
    if (replicaAssignmentStr == null || replicaAssignmentStr == "") {
      // 确定startIndex
      val startIndex = math.max(0, brokerMetadatas.indexWhere(_.id >= existingReplicaListForPartitionZero.head))
      // 对新增分区进行副本自动分配，注意fixedStartIndex和startPartitionId参数的取值
      AdminUtils.assignReplicasToBrokers(brokerMetadatas, partitionsToAdd, existingReplicaListForPartitionZero.size,
        startIndex, existingPartitionsReplicaList.size)
    }
    else
      // 解析replica-assignment参数，其中会进行一系列有效性检测
      getManualReplicaAssignment(replicaAssignmentStr, brokerMetadatas.map(_.id).toSet,
        existingPartitionsReplicaList.size, checkBrokerAvailable)

  // check if manual assignment has the right replication factor
  // 检测新增Partition的副本数是否正常
  val unmatchedRepFactorList = newPartitionReplicaList.values.filter(p => (p.size != existingReplicaListForPartitionZero.size))
  if (unmatchedRepFactorList.size != 0)
    throw new AdminOperationException("The replication factor in manual replication assignment " +
      " is not equal to the existing replication factor for the topic " + existingReplicaListForPartitionZero.size)

  info("Add partition list for %s is %s".format(topic, newPartitionReplicaList))
  // 将原有分区的新增分区的副本分配整理为集合
  val partitionReplicaList = existingPartitionsReplicaList.map(p => p._1.partition -> p._2)
  // add the new list
  // 将最终的副本分配结果写入Zookeeper
  partitionReplicaList ++= newPartitionReplicaList
  // 该方法最后一个参数表示只更新副本分配情况
  AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topic, partitionReplicaList, update = true)
}

def describeTopic(zkUtils: ZkUtils, opts: TopicCommandOptions) {
  // 从Zookeeper读取指定Topic的列表
  val topics = getTopics(zkUtils, opts)

  // 获取--under-replicated-partitions参数
  val reportUnderReplicatedPartitions = if (opts.options.has(opts.reportUnderReplicatedPartitionsOpt)) true else false
  // 获取--unavailable-partitions参数
  val reportUnavailablePartitions = if (opts.options.has(opts.reportUnavailablePartitionsOpt)) true else false
  // 获取--topics-with-overrides参数
  val reportOverriddenConfigs = if (opts.options.has(opts.topicsWithOverridesOpt)) true else false
  // 获取所有可用的Broker
  val liveBrokers = zkUtils.getAllBrokersInCluster().map(_.id).toSet

  // 遍历需要查看的Topic
  for (topic <- topics) {
    // 从Zookeeper中获取每个Topic的分区分配
    zkUtils.getPartitionAssignmentForTopics(List(topic)).get(topic) match {
      case Some(topicPartitionAssignment) => // 能够获取到
        // 处理配置
        val describeConfigs: Boolean = !reportUnavailablePartitions && !reportUnderReplicatedPartitions
        val describePartitions: Boolean = !reportOverriddenConfigs

        // 对分区分配进行排序
        val sortedPartitions = topicPartitionAssignment.toList.sortWith((m1, m2) => m1._1 < m2._1)
        if (describeConfigs) {

          // 获取Topic的配置信息
          val configs = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic, topic)

          // 没有设置--topics-with-overrides参数，且--config配置为空
          if (!reportOverriddenConfigs || configs.size() != 0) {
            // 分区数
            val numPartitions = topicPartitionAssignment.size
            // 副本因子
            val replicationFactor = topicPartitionAssignment.head._2.size
            // 打印主题信息
            println("Topic:%s\tPartitionCount:%d\tReplicationFactor:%d\tConfigs:%s"
              .format(topic, numPartitions, replicationFactor, configs.map(kv => kv._1 + "=" + kv._2).mkString(",")))
          }
        }

        // 判断是否打印分区信息
        if (describePartitions) {
          // 遍历已排序的分区
          for ((partitionId, assignedReplicas) <- sortedPartitions) {
            // ISR副本集合
            val inSyncReplicas = zkUtils.getInSyncReplicasForPartition(topic, partitionId)
            // Leader副本ID
            val leader = zkUtils.getLeaderForPartition(topic, partitionId)
            if ((!reportUnderReplicatedPartitions && !reportUnavailablePartitions) ||
                (reportUnderReplicatedPartitions && inSyncReplicas.size < assignedReplicas.size) ||
                (reportUnavailablePartitions && (!leader.isDefined || !liveBrokers.contains(leader.get)))) {
              // 打印信息
              print("\tTopic: " + topic)
              print("\tPartition: " + partitionId)
              print("\tLeader: " + (if(leader.isDefined) leader.get else "none"))
              print("\tReplicas: " + assignedReplicas.mkString(","))
              println("\tIsr: " + inSyncReplicas.mkString(","))
            }
          }
        }
      case None => // 主题不存在
        println("Topic " + topic + " doesn't exist!")
    }
  }
}

def deleteTopic(zkUtils: ZkUtils, opts: TopicCommandOptions) {
  // 获取指定的主题列表
  val topics = getTopics(zkUtils, opts)
  // --if-exists参数
  val ifExists = if (opts.options.has(opts.ifExistsOpt)) true else false

  // 根据参数进行检查
  if (topics.length == 0 && !ifExists) {
    throw new IllegalArgumentException("Topic %s does not exist on ZK path %s".format(opts.options.valueOf(opts.topicOpt),
        opts.options.valueOf(opts.zkConnectOpt)))
  }

  // 遍历指定的主题
  topics.foreach { topic =>
    try {
      if (Topic.isInternal(topic)) { // 不可删除内部分区
        throw new AdminOperationException("Topic %s is a kafka internal topic and is not allowed to be marked for deletion.".format(topic))
      } else {
        /**
          * 将待删除的Topic名称写入到Zookeeper的/admin/delete_topics路径下
          * 这将触发DeleteTopicsListener将待删除Topic交由TopicDeletionManager处理
          */
        zkUtils.createPersistentPath(getDeleteTopicPath(topic))
        println("Topic %s is marked for deletion.".format(topic))
        println("Note: This will have no impact if delete.topic.enable is not set to true.")
      }
    } catch {
      case e: ZkNodeExistsException =>
        println("Topic %s is already marked for deletion.".format(topic))
      case e: AdminOperationException =>
        throw e
      case e: Throwable =>
        throw new AdminOperationException("Error while deleting topic %s".format(topic))
    }
  }
}

def main(args: Array[String]): Unit = {
  val parser = new OptionParser
  val jsonFileOpt = parser.accepts("path-to-json-file", "The JSON file with the list of partitions " +
    "for which preferred replica leader election should be done, in the following format - \n" +
     "{\"partitions\":\n\t[{\"topic\": \"foo\", \"partition\": 1},\n\t {\"topic\": \"foobar\", \"partition\": 2}]\n}\n" +
    "Defaults to all existing partitions")
    .withRequiredArg
    .describedAs("list of partitions for which preferred replica leader election needs to be triggered")
    .ofType(classOf[String])
  val zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the " +
    "form host:port. Multiple URLS can be given to allow fail-over.")
    .withRequiredArg
    .describedAs("urls")
    .ofType(classOf[String])

  if(args.length == 0)
    CommandLineUtils.printUsageAndDie(parser, "This tool causes leadership for each partition to be transferred back to the 'preferred replica'," + 
                                              " it can be used to balance leadership among the servers.")

  // 解析及检测--path-to-json-file和--zookeeper参数
  val options = parser.parse(args : _*)

  CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt)

  // 创建Zookeeper连接
  val zkConnect = options.valueOf(zkConnectOpt)
  var zkClient: ZkClient = null
  var zkUtils: ZkUtils = null
  try {
    zkClient = ZkUtils.createZkClient(zkConnect, 30000, 30000)
    zkUtils = ZkUtils(zkConnect, 
                      30000,
                      30000,
                      JaasUtils.isZkSecurityEnabled())

    // 获取需要进行"优先副本"选举的分区集合
    val partitionsForPreferredReplicaElection =
      // 未指定--path-to-json-file参数则返回全部分区
      if (!options.has(jsonFileOpt))
        zkUtils.getAllPartitions()
      else
        // 解析--path-to-json-file参数指定的输入文件
        parsePreferredReplicaElectionData(Utils.readFileAsString(options.valueOf(jsonFileOpt)))

    // 创建PreferredReplicaLeaderElectionCommand对象
    val preferredReplicaElectionCommand = new PreferredReplicaLeaderElectionCommand(zkUtils, partitionsForPreferredReplicaElection)

    // 将指定的分区写入到Zookeeper的/admin/preferred_replica_election节点中
    preferredReplicaElectionCommand.moveLeaderToPreferredReplica()
    println("Successfully started preferred replica election for partitions %s".format(partitionsForPreferredReplicaElection))
  } catch {
    case e: Throwable =>
      println("Failed to start preferred replica election")
      println(Utils.stackTrace(e))
  } finally {
    if (zkClient != null)
      zkClient.close()
  }
}

def moveLeaderToPreferredReplica() = {
  try {
    // 检查指定的Topic是否包含指定的分区
    val validPartitions = partitions.filter(p => validatePartition(zkUtils, p.topic, p.partition))
    // 将信息写入Zookeeper
    PreferredReplicaLeaderElectionCommand.writePreferredReplicaElectionData(zkUtils, validPartitions)
  } catch {
    case e: Throwable => throw new AdminCommandFailedException("Admin command failed", e)
  }
}

def writePreferredReplicaElectionData(zkUtils: ZkUtils,
                                      partitionsUndergoingPreferredReplicaElection: scala.collection.Set[TopicAndPartition]) {
  // 得到/admin/preferred_replica_election节点的路径
  val zkPath = ZkUtils.PreferredReplicaLeaderElectionPath
  // 构造分区字典
  val partitionsList = partitionsUndergoingPreferredReplicaElection.map(e => Map("topic" -> e.topic, "partition" -> e.partition))
  // 将数据转换为JSON格式
  val jsonData = Json.encode(Map("version" -> 1, "partitions" -> partitionsList))
  try {
    // 写入Zookeeper
    zkUtils.createPersistentPath(zkPath, jsonData)
    info("Created preferred replica election path with %s".format(jsonData))
  } catch {
    case nee: ZkNodeExistsException =>
      val partitionsUndergoingPreferredReplicaElection =
        PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(zkUtils.readData(zkPath)._1)
      throw new AdminOperationException("Preferred replica leader election currently in progress for " +
        "%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection))
    case e2: Throwable => throw new AdminOperationException(e2.toString)
  }
}

def verifyAssignment(zkUtils: ZkUtils, opts: ReassignPartitionsCommandOptions) {
  // 检查--reassignment-json-file参数
  if(!opts.options.has(opts.reassignmentJsonFileOpt))
    CommandLineUtils.printUsageAndDie(opts.parser, "If --verify option is used, command must include --reassignment-json-file that was used during the --execute option")
  // 从--reassignment-json-file参数指定的JSON文件读取分配信息
  val jsonFile = opts.options.valueOf(opts.reassignmentJsonFileOpt)
  val jsonString = Utils.readFileAsString(jsonFile)

  // 解析JSON数据，主要是检查是否有重复的键
  val partitionsToBeReassigned = zkUtils.parsePartitionReassignmentData(jsonString)

  println("Status of partition reassignment:")
  // 检测每个分区的迁移情况
  val reassignedPartitionsStatus = checkIfReassignmentSucceeded(zkUtils, partitionsToBeReassigned)

  // 针对迁移情况作出打印信息
  reassignedPartitionsStatus.foreach { partition =>
    partition._2 match {
      case ReassignmentCompleted =>
        println("Reassignment of partition %s completed successfully".format(partition._1))
      case ReassignmentFailed =>
        println("Reassignment of partition %s failed".format(partition._1))
      case ReassignmentInProgress =>
        println("Reassignment of partition %s is still in progress".format(partition._1))
    }
  }
}

private def checkIfReassignmentSucceeded(zkUtils: ZkUtils, partitionsToBeReassigned: Map[TopicAndPartition, Seq[Int]])
:Map[TopicAndPartition, ReassignmentStatus] = {
  // 从Zookeeper的/admin/reassign_partitions路径下获取正在执行副本重新分配的分区信息
  val partitionsBeingReassigned = zkUtils.getPartitionsBeingReassigned().mapValues(_.newReplicas)
  // 对每个分区都调用checkIfPartitionReassignmentSucceeded()方法检测是否迁移成功
  partitionsToBeReassigned.map { topicAndPartition =>
    (topicAndPartition._1, checkIfPartitionReassignmentSucceeded(zkUtils,topicAndPartition._1,
      topicAndPartition._2, partitionsToBeReassigned, partitionsBeingReassigned))
  }
}

def checkIfPartitionReassignmentSucceeded(zkUtils: ZkUtils, topicAndPartition: TopicAndPartition,
                                          reassignedReplicas: Seq[Int],
                                          partitionsToBeReassigned: Map[TopicAndPartition, Seq[Int]],
                                          partitionsBeingReassigned: Map[TopicAndPartition, Seq[Int]]): ReassignmentStatus = {
  // 获取分区中副本重新分配的模板，若已经分配完毕，则返回空
  val newReplicas = partitionsToBeReassigned(topicAndPartition)
  partitionsBeingReassigned.get(topicAndPartition) match {
    // ZooKeeper中的/admin/reassign_partitions路径下依然有此分区信息，则表示正在进行副本重新分配
    case Some(partition) => ReassignmentInProgress
    case None =>
      // 当前AR集合与迁移目标一致，即为重新分配成功
      // check if the current replica assignment matches the expected one after reassignment
      val assignedReplicas = zkUtils.getReplicasForPartition(topicAndPartition.topic, topicAndPartition.partition)
      if(assignedReplicas == newReplicas)
        ReassignmentCompleted
      else {
        // 迁移已经完成，但当前AR集合与迁移目标不一致，表示迁移失败
        println(("ERROR: Assigned replicas (%s) don't match the list of replicas for reassignment (%s)" +
          " for partition %s").format(assignedReplicas.mkString(","), newReplicas.mkString(","), topicAndPartition))
        ReassignmentFailed
      }
  }
}

def generateAssignment(zkUtils: ZkUtils, brokerListToReassign: Seq[Int], topicsToMoveJsonString: String, disableRackAware: Boolean): (Map[TopicAndPartition, Seq[Int]], Map[TopicAndPartition, Seq[Int]]) = {
  // 获取待迁移的Topic
  val topicsToReassign = zkUtils.parseTopicsData(topicsToMoveJsonString)
  // 检测待迁移Top次集合中是否存在重复的Topic，如果存在则抛出异常
  val duplicateTopicsToReassign = CoreUtils.duplicates(topicsToReassign)
  if (duplicateTopicsToReassign.nonEmpty)
    throw new AdminCommandFailedException("List of topics to reassign contains duplicate entries: %s".format(duplicateTopicsToReassign.mkString(",")))
  // 获取待迁移Topic的当前副本分配情况
  val currentAssignment = zkUtils.getReplicaAssignmentForTopics(topicsToReassign)

  // 按照Topic名进行分组
  val groupedByTopic = currentAssignment.groupBy { case (tp, _) => tp.topic }
  // 决定是否考虑机架感知
  val rackAwareMode = if (disableRackAware) RackAwareMode.Disabled else RackAwareMode.Enforced
  // 获取--broker-list参数指定的Broker的信息，包括了机架信息
  val brokerMetadatas = AdminUtils.getBrokerMetadatas(zkUtils, rackAwareMode, Some(brokerListToReassign))

  // 记录副本迁移的目标
  val partitionsToBeReassigned = mutable.Map[TopicAndPartition, Seq[Int]]()
  // 遍历每个待迁移的Topic
  groupedByTopic.foreach { case (topic, assignment) =>
    val (_, replicas) = assignment.head
    // 进行副本自动分配
    val assignedReplicas = AdminUtils.assignReplicasToBrokers(brokerMetadatas, assignment.size, replicas.size)
    // 记录此Topic副本迁移后的结果
    partitionsToBeReassigned ++= assignedReplicas.map { case (partition, replicas) =>
      (TopicAndPartition(topic, partition) -> replicas)
    }
  }
  // 返回生产的副本迁移目标和当前副本分配的情况
  (partitionsToBeReassigned, currentAssignment)
}

def executeAssignment(zkUtils: ZkUtils,reassignmentJsonString: String) {
  // 解析--reassignment-json-file参数指定的输入文件内容并检测是否为空
  val partitionsToBeReassigned = zkUtils.parsePartitionReassignmentDataWithoutDedup(reassignmentJsonString)
  if (partitionsToBeReassigned.isEmpty)
    throw new AdminCommandFailedException("Partition reassignment data file is empty")

  // 检测待迁移Topic是否存在重复
  val duplicateReassignedPartitions = CoreUtils.duplicates(partitionsToBeReassigned.map{ case(tp,replicas) => tp})
  if (duplicateReassignedPartitions.nonEmpty)
    throw new AdminCommandFailedException("Partition reassignment contains duplicate topic partitions: %s".format(duplicateReassignedPartitions.mkString(",")))

  // 检测待迁移分区的副本分配是否存在重复
  val duplicateEntries= partitionsToBeReassigned
    .map{ case(tp,replicas) => (tp, CoreUtils.duplicates(replicas))}
    .filter{ case (tp,duplicatedReplicas) => duplicatedReplicas.nonEmpty }
  if (duplicateEntries.nonEmpty) {
    val duplicatesMsg = duplicateEntries
      .map{ case (tp,duplicateReplicas) => "%s contains multiple entries for %s".format(tp, duplicateReplicas.mkString(",")) }
      .mkString(". ")
    throw new AdminCommandFailedException("Partition replica lists may not contain duplicate entries: %s".format(duplicatesMsg))
  }

  // 创建ReassignPartitionsCommand
  val reassignPartitionsCommand = new ReassignPartitionsCommand(zkUtils, partitionsToBeReassigned.toMap)
  // before starting assignment, output the current replica assignment to facilitate rollback

  // 从Zookeeper中获取当前分区的副本分配情况并打印
  val currentPartitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(partitionsToBeReassigned.map(_._1.topic))
  println("Current partition replica assignment\n\n%s\n\nSave this to use as the --reassignment-json-file option during rollback"
    .format(zkUtils.formatAsReassignmentJson(currentPartitionReplicaAssignment)))

  // reassignPartitions()方法会将迁移的目标写入到Zookeeper的/admin/reassign_partitions路径
  // start the reassignment
  if(reassignPartitionsCommand.reassignPartitions())
    println("Successfully started reassignment of partitions %s".format(zkUtils.formatAsReassignmentJson(partitionsToBeReassigned.toMap)))
  else
    println("Failed to reassign partitions %s".format(partitionsToBeReassigned))
}

def reassignPartitions(): Boolean = {
  try {
    // 验证分区分区是否存在
    val validPartitions = partitions.filter(p => validatePartition(zkUtils, p._1.topic, p._1.partition))
    if(validPartitions.isEmpty) {
      // 没有通过验证的分区，直接返回false
      false
    }
    else {
      // 构建json数据
      val jsonReassignmentData = zkUtils.formatAsReassignmentJson(validPartitions)
      // 在Zookeeper/admin/reassign_partitions路径下写入数据
      zkUtils.createPersistentPath(ZkUtils.ReassignPartitionsPath, jsonReassignmentData)
      // 返回true
      true
    }
  } catch {
    case ze: ZkNodeExistsException =>
      val partitionsBeingReassigned = zkUtils.getPartitionsBeingReassigned()
      throw new AdminCommandFailedException("Partition reassignment currently in " +
      "progress for %s. Aborting operation".format(partitionsBeingReassigned))
    case e: Throwable => error("Admin command failed", e); false
  }
}

def main(args: Array[String]) {

  try {
    // 读取命令行参数并进行解析
    val config = new ProducerConfig(args)
    // 创建LineMessageReader对象
    val reader = Class.forName(config.readerClass).newInstance().asInstanceOf[MessageReader]

    // 初始化LineMessageReader对象
    reader.init(System.in, getReaderProps(config))

    val producer =
      if(config.useOldProducer) {
        new OldProducer(getOldProducerProps(config))
      } else {
        // 新版本的Producer
        new NewShinyProducer(getNewProducerProps(config))
      }

    // 添加JVM关闭钩子
    Runtime.getRuntime.addShutdownHook(new Thread() {
      override def run() {
        // 关闭生产者
        producer.close()
      }
    })

    var message: ProducerRecord[Array[Byte], Array[Byte]] = null
    do {
      // 读取数据
      message = reader.readMessage()
      if (message != null)
        // 发送消息
        producer.send(message.topic, message.key, message.value)
    } while (message != null)
  } catch {
    case e: joptsimple.OptionException =>
      System.err.println(e.getMessage)
      System.exit(1)
    case e: Exception =>
      e.printStackTrace
      System.exit(1)
  }
  System.exit(0)
}