kafka.py 文件源码

def broker_install(host_config):
    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)

    java.v8_install(host_config)

    software_config = helper.get_software_config(host_config, 'kafka-broker')

    version = software_config.get('version', '0.10.0.1')

    put('{}/software/scripts/kafka-broker.sh'.format(os.getcwd()), '~/', use_sudo=True)
    sudo("chmod +x kafka-broker.sh")
    sudo(". ~/kafka-broker.sh {}".format(version))

    broker_id = software_config.get('broker-id', '0')
    zk_hosts = software_config.get('zookeeper-hosts', 'localhost:2181')
    log_directories = software_config.get('log-directories', '/var/lib/kafka-logs')

    tag = '## ---- CUSTOM CONFIGURATION ---'

    sudo('echo "{}" | sudo tee -a /srv/kafka/config/server.properties'.format(tag))
    sudo('echo "delete.topic.enable = true" | sudo tee -a /srv/kafka/config/server.properties')
    sudo('echo "broker.id={}" | sudo tee -a /srv/kafka/config/server.properties'.format(broker_id))
    sudo('echo "zookeeper.connect={}" | sudo tee -a /srv/kafka/config/server.properties'.format(zk_hosts))
    sudo('echo "log.dirs={}" | sudo tee -a /srv/kafka/config/server.properties'.format(log_directories))
    sudo('echo "listeners=PLAINTEXT://{}:9093" | sudo tee -a /srv/kafka/config/server.properties'.format(host_config['private-ip']))
    sudo('echo "{}" | sudo tee -a /srv/kafka/config/server.properties'.format(tag))

    sudo("service kafka restart")