def stop(self, env, upgrade_type=None):
    import params
    env.set_params(params)
    ensure_base_directories()
    daemon_cmd = format('source {params.conf_dir}/kafka-env.sh; {params.kafka_bin} stop')
    Execute(daemon_cmd,
            user=params.kafka_user,
    )
    File (params.kafka_pid_file,
          action = "delete"
    )