kafka.py 文件源码

def create_topic(config_file, topic, replication_factor=1, partitions=1):
    """Creates a Kafka topic | args: config_file, topic name, replication factor, partitions"""
    cfg = helper.get_config(config_file)
    host_config = get_kafka_host_cfg(cfg)

    cmd = "/srv/kafka/bin/kafka-topics.sh --create --zookeeper {} " \
          "--replication-factor {} --partitions {} --topic {}".format(get_zk_host(cfg),
                                                                      replication_factor,
                                                                      partitions,
                                                                      topic)
    env.host_string = helper.get_env_host_string(host_config)
    env.user = helper.get_env_user(host_config)
    env.key_filename = helper.get_env_key_filename(host_config)

    sudo(cmd)