// 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.

#include <dsn/dist/fmt_logging.h>
#include <dsn/utility/fail_point.h>

#include "meta_bulk_load_service.h"

namespace dsn {
namespace replication {

DSN_DEFINE_uint32("meta_server",
                  bulk_load_max_rollback_times,
                  10,
                  "if bulk load rollback time "
                  "exceed this value, meta won't "
                  "rollback bulk load process to "
                  "downloading, but turn it into "
                  "failed");
DSN_TAG_VARIABLE(bulk_load_max_rollback_times, FT_MUTABLE);

DSN_DEFINE_uint32("meta_server",
                  bulk_load_ingestion_concurrent_count,
                  4,
                  "max partition_count executing ingestion at the same time");
DSN_TAG_VARIABLE(bulk_load_ingestion_concurrent_count, FT_MUTABLE);

bulk_load_service::bulk_load_service(meta_service *meta_svc, const std::string &bulk_load_dir)
    : _meta_svc(meta_svc), _state(meta_svc->get_server_state()), _bulk_load_root(bulk_load_dir)
{
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::initialize_bulk_load_service()
{
    _sync_bulk_load_storage =
        make_unique<mss::meta_storage>(_meta_svc->get_remote_storage(), &_sync_tracker);

    create_bulk_load_root_dir();
    _sync_tracker.wait_outstanding_tasks();

    try_to_continue_bulk_load();
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::on_start_bulk_load(start_bulk_load_rpc rpc)
{
    FAIL_POINT_INJECT_F("meta_on_start_bulk_load",
                        [=](dsn::string_view) { rpc.response().err = ERR_OK; });

    const auto &request = rpc.request();
    auto &response = rpc.response();
    response.err = ERR_OK;

    std::shared_ptr<app_state> app;
    {
        zauto_read_lock l(app_lock());

        app = _state->get_app(request.app_name);
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            derror_f("app({}) is not existed or not available", request.app_name);
            response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
            response.hint_msg = fmt::format(
                "app {}", response.err == ERR_APP_NOT_EXIST ? "not existed" : "dropped");
            return;
        }

        if (app->is_bulk_loading) {
            derror_f("app({}) is already executing bulk load, please wait", app->app_name);
            response.err = ERR_BUSY;
            response.hint_msg = "app is already executing bulk load";
            return;
        }
    }

    std::string hint_msg;
    error_code e = check_bulk_load_request_params(request.app_name,
                                                  request.cluster_name,
                                                  request.file_provider_type,
                                                  request.remote_root_path,
                                                  app->app_id,
                                                  app->partition_count,
                                                  hint_msg);
    if (e != ERR_OK) {
        response.err = e;
        response.hint_msg = hint_msg;
        return;
    }

    ddebug_f("app({}) start bulk load, cluster_name = {}, provider = {}",
             request.app_name,
             request.cluster_name,
             request.file_provider_type);

    // avoid possible load balancing
    _meta_svc->set_function_level(meta_function_level::fl_steady);

    tasking::enqueue(LPC_META_STATE_NORMAL,
                     _meta_svc->tracker(),
                     [this, rpc, app]() { do_start_app_bulk_load(std::move(app), std::move(rpc)); },
                     server_state::sStateHash);
}

// ThreadPool: THREAD_POOL_META_SERVER
error_code bulk_load_service::check_bulk_load_request_params(const std::string &app_name,
                                                             const std::string &cluster_name,
                                                             const std::string &file_provider,
                                                             const std::string &remote_root_path,
                                                             const int32_t app_id,
                                                             const int32_t partition_count,
                                                             std::string &hint_msg)
{
    FAIL_POINT_INJECT_F("meta_check_bulk_load_request_params",
                        [](dsn::string_view) -> error_code { return ERR_OK; });

    // check file provider
    dsn::dist::block_service::block_filesystem *blk_fs =
        _meta_svc->get_block_service_manager().get_or_create_block_filesystem(file_provider);
    if (blk_fs == nullptr) {
        derror_f("invalid remote file provider type: {}", file_provider);
        hint_msg = "invalid file_provider";
        return ERR_INVALID_PARAMETERS;
    }

    // sync get bulk_load_info file_handler
    const std::string remote_path =
        get_bulk_load_info_path(app_name, cluster_name, remote_root_path);
    dsn::dist::block_service::create_file_request cf_req;
    cf_req.file_name = remote_path;
    cf_req.ignore_metadata = true;
    error_code err = ERR_OK;
    dsn::dist::block_service::block_file_ptr file_handler = nullptr;
    blk_fs
        ->create_file(
            cf_req,
            TASK_CODE_EXEC_INLINED,
            [&err, &file_handler](const dsn::dist::block_service::create_file_response &resp) {
                err = resp.err;
                file_handler = resp.file_handle;
            })
        ->wait();
    if (err != ERR_OK || file_handler == nullptr) {
        derror_f(
            "failed to get file({}) handler on remote provider({})", remote_path, file_provider);
        hint_msg = "file_provider error";
        return ERR_FILE_OPERATION_FAILED;
    }

    // sync read bulk_load_info on file provider
    dsn::dist::block_service::read_response r_resp;
    file_handler
        ->read(dsn::dist::block_service::read_request{0, -1},
               TASK_CODE_EXEC_INLINED,
               [&r_resp](const dsn::dist::block_service::read_response &resp) { r_resp = resp; })
        ->wait();
    if (r_resp.err != ERR_OK) {
        derror_f("failed to read file({}) on remote provider({}), error = {}",
                 file_provider,
                 remote_path,
                 r_resp.err.to_string());
        hint_msg = "read bulk_load_info failed";
        return r_resp.err;
    }

    bulk_load_info bl_info;
    if (!::dsn::json::json_forwarder<bulk_load_info>::decode(r_resp.buffer, bl_info)) {
        derror_f("file({}) is damaged on remote file provider({})", remote_path, file_provider);
        hint_msg = "bulk_load_info damaged";
        return ERR_CORRUPTION;
    }

    if (bl_info.app_id != app_id || bl_info.partition_count != partition_count) {
        derror_f("app({}) information is inconsistent, local app_id({}) VS remote app_id({}), "
                 "local partition_count({}) VS remote partition_count({})",
                 app_name,
                 app_id,
                 bl_info.app_id,
                 partition_count,
                 bl_info.partition_count);
        hint_msg = "app_id or partition_count is inconsistent";
        return ERR_INCONSISTENT_STATE;
    }

    return ERR_OK;
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::do_start_app_bulk_load(std::shared_ptr<app_state> app,
                                               start_bulk_load_rpc rpc)
{
    app_info info = *app;
    info.__set_is_bulk_loading(true);

    blob value = dsn::json::json_forwarder<app_info>::encode(info);
    _meta_svc->get_meta_storage()->set_data(
        _state->get_app_path(*app), std::move(value), [app, rpc, this]() {
            {
                zauto_write_lock l(app_lock());
                app->is_bulk_loading = true;
            }
            {
                zauto_write_lock l(_lock);
                _bulk_load_app_id.insert(app->app_id);
                _apps_in_progress_count[app->app_id] = app->partition_count;
                _apps_ingesting_count[app->app_id] = 0;
            }
            create_app_bulk_load_dir(
                app->app_name, app->app_id, app->partition_count, std::move(rpc));
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::create_app_bulk_load_dir(const std::string &app_name,
                                                 int32_t app_id,
                                                 int32_t partition_count,
                                                 start_bulk_load_rpc rpc)
{
    const auto &req = rpc.request();

    app_bulk_load_info ainfo;
    ainfo.app_id = app_id;
    ainfo.app_name = app_name;
    ainfo.partition_count = partition_count;
    ainfo.status = bulk_load_status::BLS_DOWNLOADING;
    ainfo.cluster_name = req.cluster_name;
    ainfo.file_provider_type = req.file_provider_type;
    ainfo.remote_root_path = req.remote_root_path;
    blob value = dsn::json::json_forwarder<app_bulk_load_info>::encode(ainfo);

    _meta_svc->get_meta_storage()->create_node(
        get_app_bulk_load_path(app_id), std::move(value), [rpc, ainfo, this]() {
            dinfo_f("create app({}) bulk load dir", ainfo.app_name);
            {
                zauto_write_lock l(_lock);
                _app_bulk_load_info[ainfo.app_id] = ainfo;
                _apps_pending_sync_flag[ainfo.app_id] = false;
                _apps_rollback_count[ainfo.app_id] = 0;
            }
            for (int32_t i = 0; i < ainfo.partition_count; ++i) {
                create_partition_bulk_load_dir(
                    ainfo.app_name, gpid(ainfo.app_id, i), ainfo.partition_count, std::move(rpc));
            }
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::create_partition_bulk_load_dir(const std::string &app_name,
                                                       const gpid &pid,
                                                       int32_t partition_count,
                                                       start_bulk_load_rpc rpc)
{
    partition_bulk_load_info pinfo;
    pinfo.status = bulk_load_status::BLS_DOWNLOADING;
    blob value = dsn::json::json_forwarder<partition_bulk_load_info>::encode(pinfo);

    _meta_svc->get_meta_storage()->create_node(
        get_partition_bulk_load_path(pid),
        std::move(value),
        [app_name, pid, partition_count, rpc, pinfo, this]() {
            dinfo_f("app({}) create partition({}) bulk_load_info", app_name, pid.to_string());
            {
                zauto_write_lock l(_lock);
                _partition_bulk_load_info[pid] = pinfo;
                _partitions_pending_sync_flag[pid] = false;
                if (--_apps_in_progress_count[pid.get_app_id()] == 0) {
                    ddebug_f("app({}) start bulk load succeed", app_name);
                    _apps_in_progress_count[pid.get_app_id()] = partition_count;
                    rpc.response().err = ERR_OK;
                }
            }
            // start send bulk load to replica servers
            partition_bulk_load(app_name, pid);
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::partition_bulk_load(const std::string &app_name, const gpid &pid)
{
    FAIL_POINT_INJECT_F("meta_bulk_load_partition_bulk_load", [](dsn::string_view) {});

    rpc_address primary_addr;
    ballot b;
    {
        zauto_read_lock l(app_lock());
        std::shared_ptr<app_state> app = _state->get_app(pid.get_app_id());
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            dwarn_f("app(name={}, id={}) is not existed, set bulk load failed",
                    app_name,
                    pid.get_app_id());
            handle_app_unavailable(pid.get_app_id(), app_name);
            return;
        }
        primary_addr = app->partitions[pid.get_partition_index()].primary;
        b = app->partitions[pid.get_partition_index()].ballot;
    }

    if (primary_addr.is_invalid()) {
        dwarn_f("app({}) partition({}) primary is invalid, try it later", app_name, pid);
        tasking::enqueue(LPC_META_STATE_NORMAL,
                         _meta_svc->tracker(),
                         std::bind(&bulk_load_service::partition_bulk_load, this, app_name, pid),
                         0,
                         std::chrono::seconds(1));
        return;
    }

    zauto_read_lock l(_lock);
    const app_bulk_load_info &ainfo = _app_bulk_load_info[pid.get_app_id()];
    auto req = make_unique<bulk_load_request>();
    req->pid = pid;
    req->app_name = app_name;
    req->primary_addr = primary_addr;
    req->remote_provider_name = ainfo.file_provider_type;
    req->cluster_name = ainfo.cluster_name;
    req->meta_bulk_load_status = get_partition_bulk_load_status_unlocked(pid);
    req->ballot = b;
    req->query_bulk_load_metadata = is_partition_metadata_not_updated_unlocked(pid);
    req->remote_root_path = ainfo.remote_root_path;

    ddebug_f("send bulk load request to node({}), app({}), partition({}), partition "
             "status = {}, remote provider = {}, cluster_name = {}, remote_root_path = {}",
             primary_addr.to_string(),
             app_name,
             pid,
             dsn::enum_to_string(req->meta_bulk_load_status),
             req->remote_provider_name,
             req->cluster_name,
             req->remote_root_path);

    bulk_load_rpc rpc(std::move(req), RPC_BULK_LOAD, 0_ms, 0, pid.thread_hash());
    rpc.call(primary_addr, _meta_svc->tracker(), [this, rpc](error_code err) mutable {
        on_partition_bulk_load_reply(err, rpc.request(), rpc.response());
    });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::on_partition_bulk_load_reply(error_code err,
                                                     const bulk_load_request &request,
                                                     const bulk_load_response &response)
{
    const std::string &app_name = request.app_name;
    const gpid &pid = request.pid;
    const rpc_address &primary_addr = request.primary_addr;
    int32_t interval = bulk_load_constant::BULK_LOAD_REQUEST_INTERVAL;

    if (err != ERR_OK) {
        derror_f("app({}), partition({}) failed to receive bulk load response, error = {}",
                 app_name,
                 pid,
                 err.to_string());
        try_rollback_to_downloading(app_name, pid);
        try_resend_bulk_load_request(app_name, pid, interval);
        return;
    }

    if (response.err == ERR_OBJECT_NOT_FOUND || response.err == ERR_INVALID_STATE) {
        derror_f(
            "app({}), partition({}) doesn't exist or has invalid state on node({}), error = {}",
            app_name,
            pid,
            primary_addr.to_string(),
            response.err.to_string());
        try_rollback_to_downloading(app_name, pid);
        try_resend_bulk_load_request(app_name, pid, interval);
        return;
    }

    if (response.err == ERR_BUSY) {
        dwarn_f("node({}) has enough replicas downloading, wait for next round to send bulk load "
                "request for app({}), partition({})",
                primary_addr.to_string(),
                app_name,
                pid);
        try_resend_bulk_load_request(app_name, pid, interval);
        return;
    }

    if (response.err != ERR_OK) {
        derror_f("app({}), partition({}) handle bulk load response failed, error = {}, primary "
                 "status = {}",
                 app_name,
                 pid,
                 response.err.to_string(),
                 dsn::enum_to_string(response.primary_bulk_load_status));
        handle_bulk_load_failed(pid.get_app_id());
        try_resend_bulk_load_request(app_name, pid, interval);
        return;
    }

    // response.err = ERR_OK
    ballot current_ballot;
    {
        zauto_read_lock l(app_lock());
        std::shared_ptr<app_state> app = _state->get_app(pid.get_app_id());
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            dwarn_f("app(name={}, id={}) is not existed, set bulk load failed",
                    app_name,
                    pid.get_app_id());
            handle_app_unavailable(pid.get_app_id(), app_name);
            return;
        }
        current_ballot = app->partitions[pid.get_partition_index()].ballot;
    }
    if (request.ballot < current_ballot) {
        dwarn_f("receive out-date response, app({}), partition({}), request ballot = {}, "
                "current ballot= {}",
                app_name,
                pid,
                request.ballot,
                current_ballot);
        try_rollback_to_downloading(app_name, pid);
        try_resend_bulk_load_request(app_name, pid, interval);
        return;
    }

    // handle bulk load states reported from primary replica
    bulk_load_status::type app_status = get_app_bulk_load_status(response.pid.get_app_id());
    switch (app_status) {
    case bulk_load_status::BLS_DOWNLOADING:
        handle_app_downloading(response, primary_addr);
        break;
    case bulk_load_status::BLS_DOWNLOADED:
        update_partition_status_on_remote_storage(
            response.app_name, response.pid, bulk_load_status::BLS_INGESTING);
        // when app status is downloaded or ingesting, send request frequently
        interval = bulk_load_constant::BULK_LOAD_REQUEST_SHORT_INTERVAL;
        break;
    case bulk_load_status::BLS_INGESTING:
        handle_app_ingestion(response, primary_addr);
        interval = bulk_load_constant::BULK_LOAD_REQUEST_SHORT_INTERVAL;
        break;
    case bulk_load_status::BLS_SUCCEED:
    case bulk_load_status::BLS_FAILED:
    case bulk_load_status::BLS_CANCELED:
        handle_bulk_load_finish(response, primary_addr);
        break;
    case bulk_load_status::BLS_PAUSING:
        handle_app_pausing(response, primary_addr);
        break;
    case bulk_load_status::BLS_PAUSED:
        // paused not send request to replica servers
        return;
    default:
        // do nothing in other status
        break;
    }

    try_resend_bulk_load_request(app_name, pid, interval);
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::try_resend_bulk_load_request(const std::string &app_name,
                                                     const gpid &pid,
                                                     const int32_t interval)
{
    FAIL_POINT_INJECT_F("meta_bulk_load_resend_request", [](dsn::string_view) {});
    zauto_read_lock l(_lock);
    if (is_app_bulk_loading_unlocked(pid.get_app_id())) {
        tasking::enqueue(LPC_META_STATE_NORMAL,
                         _meta_svc->tracker(),
                         std::bind(&bulk_load_service::partition_bulk_load, this, app_name, pid),
                         0,
                         std::chrono::seconds(interval));
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_app_downloading(const bulk_load_response &response,
                                               const rpc_address &primary_addr)
{
    const std::string &app_name = response.app_name;
    const gpid &pid = response.pid;

    if (!response.__isset.total_download_progress) {
        dwarn_f(
            "receive bulk load response from node({}) app({}), partition({}), primary_status({}), "
            "but total_download_progress is not set",
            primary_addr.to_string(),
            app_name,
            pid,
            dsn::enum_to_string(response.primary_bulk_load_status));
        return;
    }

    for (const auto &kv : response.group_bulk_load_state) {
        const auto &bulk_load_states = kv.second;
        if (!bulk_load_states.__isset.download_progress ||
            !bulk_load_states.__isset.download_status) {
            dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
                    "primary_status({}), but node({}) progress or status is not set",
                    primary_addr.to_string(),
                    app_name,
                    pid,
                    dsn::enum_to_string(response.primary_bulk_load_status),
                    kv.first.to_string());
            return;
        }
        // check partition download status
        if (bulk_load_states.download_status != ERR_OK) {
            derror_f("app({}) partition({}) on node({}) meet unrecoverable error during "
                     "downloading files, error = {}",
                     app_name,
                     pid,
                     kv.first.to_string(),
                     bulk_load_states.download_status);
            handle_bulk_load_failed(pid.get_app_id());
            return;
        }
    }

    // if replica report metadata, update metadata on remote storage
    if (response.__isset.metadata && is_partition_metadata_not_updated(pid)) {
        update_partition_metadata_on_remote_storage(app_name, pid, response.metadata);
    }

    // update download progress
    int32_t total_progress = response.total_download_progress;
    ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
             "total_download_progress = {}",
             primary_addr.to_string(),
             app_name,
             pid,
             dsn::enum_to_string(response.primary_bulk_load_status),
             total_progress);
    {
        zauto_write_lock l(_lock);
        _partitions_total_download_progress[pid] = total_progress;
        _partitions_bulk_load_state[pid] = response.group_bulk_load_state;
    }

    // update partition status to `downloaded` if all replica downloaded
    if (total_progress >= bulk_load_constant::PROGRESS_FINISHED) {
        ddebug_f(
            "app({}) partirion({}) download all files from remote provider succeed", app_name, pid);
        update_partition_status_on_remote_storage(app_name, pid, bulk_load_status::BLS_DOWNLOADED);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_app_ingestion(const bulk_load_response &response,
                                             const rpc_address &primary_addr)
{
    const std::string &app_name = response.app_name;
    const gpid &pid = response.pid;

    if (!response.__isset.is_group_ingestion_finished) {
        dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
                "primary_status({}), but is_group_ingestion_finished is not set",
                primary_addr.to_string(),
                app_name,
                pid,
                dsn::enum_to_string(response.primary_bulk_load_status));
        return;
    }

    for (const auto &kv : response.group_bulk_load_state) {
        const auto &bulk_load_states = kv.second;
        if (!bulk_load_states.__isset.ingest_status) {
            dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
                    "primary_status({}), but node({}) ingestion_status is not set",
                    primary_addr.to_string(),
                    app_name,
                    pid,
                    dsn::enum_to_string(response.primary_bulk_load_status),
                    kv.first.to_string());
            return;
        }

        if (bulk_load_states.ingest_status == ingestion_status::IS_FAILED) {
            derror_f("app({}) partition({}) on node({}) ingestion failed",
                     app_name,
                     pid,
                     kv.first.to_string());
            decrease_app_ingestion_count(pid);
            handle_bulk_load_failed(pid.get_app_id());
            return;
        }
    }

    ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary_status({}), "
             "is_group_ingestion_finished = {}",
             primary_addr.to_string(),
             app_name,
             pid,
             dsn::enum_to_string(response.primary_bulk_load_status),
             response.is_group_ingestion_finished);
    {
        zauto_write_lock l(_lock);
        _partitions_bulk_load_state[pid] = response.group_bulk_load_state;
    }

    if (response.is_group_ingestion_finished) {
        ddebug_f("app({}) partition({}) ingestion files succeed", app_name, pid);
        decrease_app_ingestion_count(pid);
        update_partition_status_on_remote_storage(app_name, pid, bulk_load_status::BLS_SUCCEED);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_bulk_load_finish(const bulk_load_response &response,
                                                const rpc_address &primary_addr)
{
    const std::string &app_name = response.app_name;
    const gpid &pid = response.pid;

    if (!response.__isset.is_group_bulk_load_context_cleaned_up) {
        dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
                "primary_status({}), but is_group_bulk_load_context_cleaned_up is not set",
                primary_addr.to_string(),
                app_name,
                pid,
                dsn::enum_to_string(response.primary_bulk_load_status));
        return;
    }

    for (const auto &kv : response.group_bulk_load_state) {
        if (!kv.second.__isset.is_cleaned_up) {
            dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
                    "primary_status({}), but node({}) is_cleaned_up is not set",
                    primary_addr.to_string(),
                    app_name,
                    pid,
                    dsn::enum_to_string(response.primary_bulk_load_status),
                    kv.first.to_string());
            return;
        }
    }

    {
        zauto_read_lock l(_lock);
        if (_partitions_cleaned_up[pid]) {
            dwarn_f(
                "receive bulk load response from node({}) app({}) partition({}), current partition "
                "has already been cleaned up",
                primary_addr.to_string(),
                app_name,
                pid);
            return;
        }
    }

    // The replicas have cleaned up their bulk load states and removed temporary sst files
    bool group_cleaned_up = response.is_group_bulk_load_context_cleaned_up;
    ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
             "is_group_bulk_load_context_cleaned_up = {}",
             primary_addr.to_string(),
             app_name,
             pid,
             dsn::enum_to_string(response.primary_bulk_load_status),
             group_cleaned_up);
    {
        zauto_write_lock l(_lock);
        _partitions_cleaned_up[pid] = group_cleaned_up;
        _partitions_bulk_load_state[pid] = response.group_bulk_load_state;
    }

    if (group_cleaned_up) {
        int32_t count = 0;
        {
            zauto_write_lock l(_lock);
            count = --_apps_in_progress_count[pid.get_app_id()];
        }
        if (count == 0) {
            std::shared_ptr<app_state> app;
            {
                zauto_read_lock l(app_lock());
                app = _state->get_app(pid.get_app_id());
                if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
                    dwarn_f("app(name={}, id={}) is not existed, remove bulk load dir on remote "
                            "storage",
                            app_name,
                            pid.get_app_id());
                    remove_bulk_load_dir_on_remote_storage(pid.get_app_id(), app_name);
                    return;
                }
            }
            ddebug_f("app({}) all partitions cleanup bulk load context", app_name);
            remove_bulk_load_dir_on_remote_storage(std::move(app), true);
        }
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_app_pausing(const bulk_load_response &response,
                                           const rpc_address &primary_addr)
{
    const std::string &app_name = response.app_name;
    const gpid &pid = response.pid;

    if (!response.__isset.is_group_bulk_load_paused) {
        dwarn_f("receive bulk load response from node({}) app({}) partition({}), "
                "primary_status({}), but is_group_bulk_load_paused is not set",
                primary_addr.to_string(),
                app_name,
                pid,
                dsn::enum_to_string(response.primary_bulk_load_status));
        return;
    }

    for (const auto &kv : response.group_bulk_load_state) {
        if (!kv.second.__isset.is_paused) {
            dwarn_f("receive bulk load response from node({}) app({}), partition({}), "
                    "primary_status({}), but node({}) is_paused is not set",
                    primary_addr.to_string(),
                    app_name,
                    pid,
                    dsn::enum_to_string(response.primary_bulk_load_status),
                    kv.first.to_string());
            return;
        }
    }

    bool is_group_paused = response.is_group_bulk_load_paused;
    ddebug_f("receive bulk load response from node({}) app({}) partition({}), primary status = {}, "
             "is_group_bulk_load_paused = {}",
             primary_addr.to_string(),
             app_name,
             pid,
             dsn::enum_to_string(response.primary_bulk_load_status),
             is_group_paused);
    {
        zauto_write_lock l(_lock);
        _partitions_bulk_load_state[pid] = response.group_bulk_load_state;
    }

    if (is_group_paused) {
        ddebug_f("app({}) partirion({}) pause bulk load succeed", response.app_name, pid);
        update_partition_status_on_remote_storage(
            response.app_name, pid, bulk_load_status::BLS_PAUSED);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::try_rollback_to_downloading(const std::string &app_name, const gpid &pid)
{
    zauto_write_lock l(_lock);

    const auto app_status = get_app_bulk_load_status_unlocked(pid.get_app_id());
    if (app_status != bulk_load_status::BLS_DOWNLOADING &&
        app_status != bulk_load_status::BLS_DOWNLOADED &&
        app_status != bulk_load_status::BLS_INGESTING &&
        app_status != bulk_load_status::BLS_SUCCEED) {
        ddebug_f("app({}) status={}, no need to rollback to downloading, wait for next round",
                 app_name,
                 dsn::enum_to_string(app_status));
        return;
    }

    if (_apps_rolling_back[pid.get_app_id()]) {
        dwarn_f("app({}) is rolling back to downloading, ignore this request", app_name);
        return;
    }
    if (_apps_rollback_count[pid.get_app_id()] >= FLAGS_bulk_load_max_rollback_times) {
        dwarn_f(
            "app({}) has been rollback to downloading for {} times, make bulk load process failed",
            app_name,
            _apps_rollback_count[pid.get_app_id()]);
        update_app_status_on_remote_storage_unlocked(pid.get_app_id(),
                                                     bulk_load_status::type::BLS_FAILED);
        return;
    }
    ddebug_f("app({}) will rolling back from {} to {}, current rollback_count = {}",
             app_name,
             dsn::enum_to_string(app_status),
             dsn::enum_to_string(bulk_load_status::BLS_DOWNLOADING),
             _apps_rollback_count[pid.get_app_id()]);
    _apps_rolling_back[pid.get_app_id()] = true;
    _apps_rollback_count[pid.get_app_id()]++;
    update_app_status_on_remote_storage_unlocked(pid.get_app_id(),
                                                 bulk_load_status::type::BLS_DOWNLOADING);
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_bulk_load_failed(int32_t app_id)
{
    zauto_write_lock l(_lock);
    if (!_apps_cleaning_up[app_id]) {
        _apps_cleaning_up[app_id] = true;
        update_app_status_on_remote_storage_unlocked(app_id, bulk_load_status::BLS_FAILED);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::handle_app_unavailable(int32_t app_id, const std::string &app_name)
{
    zauto_write_lock l(_lock);
    if (is_app_bulk_loading_unlocked(app_id) && !_apps_cleaning_up[app_id]) {
        _apps_cleaning_up[app_id] = true;
        remove_bulk_load_dir_on_remote_storage(app_id, app_name);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_partition_metadata_on_remote_storage(
    const std::string &app_name, const gpid &pid, const bulk_load_metadata &metadata)
{
    zauto_read_lock l(_lock);
    partition_bulk_load_info pinfo = _partition_bulk_load_info[pid];
    pinfo.metadata = metadata;
    blob value = json::json_forwarder<partition_bulk_load_info>::encode(pinfo);

    _meta_svc->get_meta_storage()->set_data(
        get_partition_bulk_load_path(pid), std::move(value), [this, app_name, pid, pinfo]() {
            zauto_write_lock l(_lock);
            _partition_bulk_load_info[pid] = pinfo;
            ddebug_f(
                "app({}) update partition({}) bulk load metadata, file count = {}, file size = {}",
                app_name,
                pid,
                pinfo.metadata.files.size(),
                pinfo.metadata.file_total_size);
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_partition_status_on_remote_storage(const std::string &app_name,
                                                                  const gpid &pid,
                                                                  bulk_load_status::type new_status,
                                                                  bool should_send_request)
{
    zauto_write_lock l(_lock);
    partition_bulk_load_info pinfo = _partition_bulk_load_info[pid];
    if (pinfo.status == new_status && new_status != bulk_load_status::BLS_DOWNLOADING) {
        dwarn_f("app({}) partition({}) old status:{} VS new status:{}, ignore it",
                app_name,
                pid,
                dsn::enum_to_string(pinfo.status),
                dsn::enum_to_string(new_status));
        return;
    }

    if (_partitions_pending_sync_flag[pid]) {
        if (_apps_rolling_back[pid.get_app_id()] &&
            new_status == bulk_load_status::BLS_DOWNLOADING) {
            dwarn_f("app({}) partition({}) has already sync bulk load status, current_status = {}, "
                    "wait and retry to set status as {}",
                    app_name,
                    pid,
                    dsn::enum_to_string(pinfo.status),
                    dsn::enum_to_string(new_status));
            tasking::enqueue(
                LPC_META_STATE_NORMAL,
                _meta_svc->tracker(),
                std::bind(&bulk_load_service::update_partition_status_on_remote_storage,
                          this,
                          app_name,
                          pid,
                          new_status,
                          should_send_request),
                0,
                std::chrono::seconds(1));
        } else {
            ddebug_f("app({}) partition({}) has already sync bulk load status, current_status = "
                     "{}, new_status = {}, wait for next round",
                     app_name,
                     pid,
                     dsn::enum_to_string(pinfo.status),
                     dsn::enum_to_string(new_status));
        }
        return;
    }

    _partitions_pending_sync_flag[pid] = true;
    pinfo.status = new_status;
    blob value = json::json_forwarder<partition_bulk_load_info>::encode(pinfo);

    _meta_svc->get_meta_storage()->set_data(
        get_partition_bulk_load_path(pid),
        std::move(value),
        std::bind(&bulk_load_service::update_partition_status_on_remote_storage_reply,
                  this,
                  app_name,
                  pid,
                  new_status,
                  should_send_request));
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_partition_status_on_remote_storage_reply(
    const std::string &app_name,
    const gpid &pid,
    bulk_load_status::type new_status,
    bool should_send_request)
{
    {
        zauto_write_lock l(_lock);
        auto old_status = _partition_bulk_load_info[pid].status;
        _partition_bulk_load_info[pid].status = new_status;
        _partitions_pending_sync_flag[pid] = false;

        ddebug_f("app({}) update partition({}) status from {} to {}",
                 app_name,
                 pid,
                 dsn::enum_to_string(old_status),
                 dsn::enum_to_string(new_status));

        switch (new_status) {
        case bulk_load_status::BLS_DOWNLOADED:
        case bulk_load_status::BLS_INGESTING:
        case bulk_load_status::BLS_SUCCEED:
        case bulk_load_status::BLS_PAUSED:
            if (old_status != new_status && !_apps_rolling_back[pid.get_app_id()] &&
                --_apps_in_progress_count[pid.get_app_id()] == 0) {
                update_app_status_on_remote_storage_unlocked(pid.get_app_id(), new_status);
            }
            break;
        case bulk_load_status::BLS_DOWNLOADING: {
            _partitions_bulk_load_state.erase(pid);
            _partitions_total_download_progress[pid] = 0;
            _partitions_cleaned_up[pid] = false;

            if (--_apps_in_progress_count[pid.get_app_id()] == 0) {
                _apps_in_progress_count[pid.get_app_id()] =
                    _app_bulk_load_info[pid.get_app_id()].partition_count;
                _apps_rolling_back[pid.get_app_id()] = false;
                ddebug_f("app({}) restart to bulk load", app_name);
            }
        } break;
        default:
            // do nothing in other status
            break;
        }
    }
    if (should_send_request) {
        partition_bulk_load(app_name, pid);
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_app_status_on_remote_storage_unlocked(
    int32_t app_id, bulk_load_status::type new_status, bool should_send_request)
{
    FAIL_POINT_INJECT_F("meta_update_app_status_on_remote_storage_unlocked",
                        [](dsn::string_view) {});

    app_bulk_load_info ainfo = _app_bulk_load_info[app_id];
    auto old_status = ainfo.status;

    if (old_status == new_status && new_status != bulk_load_status::BLS_DOWNLOADING) {
        dwarn_f("app({}) old status:{} VS new status:{}, ignore it",
                ainfo.app_name,
                dsn::enum_to_string(old_status),
                dsn::enum_to_string(new_status));
        return;
    }

    if (_apps_pending_sync_flag[app_id]) {
        ddebug_f("app({}) has already sync bulk load status, wait and retry, current status = {}, "
                 "new status = {}",
                 ainfo.app_name,
                 dsn::enum_to_string(old_status),
                 dsn::enum_to_string(new_status));
        tasking::enqueue(LPC_META_STATE_NORMAL,
                         _meta_svc->tracker(),
                         std::bind(&bulk_load_service::update_app_status_on_remote_storage_unlocked,
                                   this,
                                   app_id,
                                   new_status,
                                   should_send_request),
                         0,
                         std::chrono::seconds(1));
        return;
    }

    _apps_pending_sync_flag[app_id] = true;
    ainfo.status = new_status;
    blob value = dsn::json::json_forwarder<app_bulk_load_info>::encode(ainfo);

    _meta_svc->get_meta_storage()->set_data(
        get_app_bulk_load_path(app_id),
        std::move(value),
        std::bind(&bulk_load_service::update_app_status_on_remote_storage_reply,
                  this,
                  ainfo,
                  old_status,
                  new_status,
                  should_send_request));
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_app_status_on_remote_storage_reply(const app_bulk_load_info &ainfo,
                                                                  bulk_load_status::type old_status,
                                                                  bulk_load_status::type new_status,
                                                                  bool should_send_request)
{
    int32_t app_id = ainfo.app_id;
    int32_t partition_count = ainfo.partition_count;
    {
        zauto_write_lock l(_lock);
        _app_bulk_load_info[app_id] = ainfo;
        _apps_pending_sync_flag[app_id] = false;
        _apps_in_progress_count[app_id] = partition_count;
        // when rollback from ingesting, ingesting_count should be reset
        if (old_status == bulk_load_status::BLS_INGESTING &&
            new_status == bulk_load_status::BLS_DOWNLOADING) {
            _apps_ingesting_count[app_id] = 0;
        }
    }

    ddebug_f("update app({}) status from {} to {}",
             ainfo.app_name,
             dsn::enum_to_string(old_status),
             dsn::enum_to_string(new_status));

    if (new_status == bulk_load_status::BLS_INGESTING) {
        for (int i = 0; i < partition_count; ++i) {
            partition_ingestion(ainfo.app_name, gpid(app_id, i));
        }
    }

    if (new_status == bulk_load_status::BLS_PAUSING ||
        new_status == bulk_load_status::BLS_DOWNLOADING ||
        new_status == bulk_load_status::BLS_CANCELED ||
        new_status == bulk_load_status::BLS_FAILED) {
        for (int i = 0; i < ainfo.partition_count; ++i) {
            update_partition_status_on_remote_storage(
                ainfo.app_name, gpid(app_id, i), new_status, should_send_request);
        }
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::partition_ingestion(const std::string &app_name, const gpid &pid)
{
    FAIL_POINT_INJECT_F("meta_bulk_load_partition_ingestion", [=](dsn::string_view) {
        if (_apps_ingesting_count[pid.get_app_id()] < FLAGS_bulk_load_ingestion_concurrent_count) {
            _apps_ingesting_count[pid.get_app_id()]++;
        }
    });

    {
        zauto_read_lock l(_lock);
        if (_apps_ingesting_count[pid.get_app_id()] >= FLAGS_bulk_load_ingestion_concurrent_count) {
            dwarn_f("app({}) has already {} partitions executing ingestion, partition({}) will "
                    "wait and try it later",
                    app_name,
                    _apps_ingesting_count[pid.get_app_id()],
                    pid);
            tasking::enqueue(
                LPC_META_STATE_NORMAL,
                _meta_svc->tracker(),
                std::bind(&bulk_load_service::partition_ingestion, this, app_name, pid),
                pid.thread_hash(),
                std::chrono::seconds(5));
            return;
        }
    }

    auto app_status = get_app_bulk_load_status(pid.get_app_id());
    if (app_status != bulk_load_status::BLS_INGESTING) {
        dwarn_f("app({}) current status is {}, partition({}), ignore it",
                app_name,
                dsn::enum_to_string(app_status),
                pid);
        return;
    }

    rpc_address primary_addr;
    {
        zauto_read_lock l(app_lock());
        std::shared_ptr<app_state> app = _state->get_app(pid.get_app_id());
        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            dwarn_f("app(name={}, id={}) is not existed, set bulk load failed",
                    app_name,
                    pid.get_app_id());
            handle_app_unavailable(pid.get_app_id(), app_name);
            return;
        }
        primary_addr = app->partitions[pid.get_partition_index()].primary;
    }

    if (primary_addr.is_invalid()) {
        dwarn_f("app({}) partition({}) primary is invalid, try it later", app_name, pid);
        tasking::enqueue(LPC_META_STATE_NORMAL,
                         _meta_svc->tracker(),
                         std::bind(&bulk_load_service::partition_ingestion, this, app_name, pid),
                         pid.thread_hash(),
                         std::chrono::seconds(1));
        return;
    }

    if (is_partition_metadata_not_updated(pid)) {
        derror_f("app({}) partition({}) doesn't have bulk load metadata, set bulk load failed",
                 app_name,
                 pid);
        handle_bulk_load_failed(pid.get_app_id());
        return;
    }

    tasking::enqueue(
        LPC_BULK_LOAD_INGESTION,
        _meta_svc->tracker(),
        std::bind(&bulk_load_service::send_ingestion_request, this, app_name, pid, primary_addr));
    {
        zauto_write_lock l(_lock);
        _apps_ingesting_count[pid.get_app_id()]++;
        ddebug_f("send ingest_request to node({}), app({}) partition({}), ingestion_count({})",
                 primary_addr.to_string(),
                 app_name,
                 pid,
                 _apps_ingesting_count[pid.get_app_id()]);
    }
}

// ThreadPool: THREAD_POOL_DEFAULT
void bulk_load_service::send_ingestion_request(const std::string &app_name,
                                               const gpid &pid,
                                               const rpc_address &primary_addr)
{
    ingestion_request req;
    req.app_name = app_name;
    {
        zauto_read_lock l(_lock);
        req.metadata = _partition_bulk_load_info[pid].metadata;
    }
    // create a client request, whose gpid field in header should be pid
    message_ex *msg = message_ex::create_request(dsn::apps::RPC_RRDB_RRDB_BULK_LOAD,
                                                 0,
                                                 pid.thread_hash(),
                                                 static_cast<uint64_t>(pid.get_partition_index()));
    auto &hdr = *msg->header;
    hdr.gpid = pid;
    dsn::marshall(msg, req);
    dsn::rpc_response_task_ptr rpc_callback = rpc::create_rpc_response_task(
        msg,
        _meta_svc->tracker(),
        [this, app_name, pid](error_code err, ingestion_response &&resp) {
            on_partition_ingestion_reply(err, std::move(resp), app_name, pid);
        });
    _meta_svc->send_request(msg, primary_addr, rpc_callback);
}

// ThreadPool: THREAD_POOL_DEFAULT
void bulk_load_service::on_partition_ingestion_reply(error_code err,
                                                     const ingestion_response &&resp,
                                                     const std::string &app_name,
                                                     const gpid &pid)
{
    if (err != ERR_OK || resp.err != ERR_OK || resp.rocksdb_error != ERR_OK) {
        decrease_app_ingestion_count(pid);
    }

    if (err == ERR_NO_NEED_OPERATE) {
        dwarn_f(
            "app({}) partition({}) has already executing ingestion, ignore this repeated request",
            app_name,
            pid);
        return;
    }

    // if meet 2pc error, ingesting will rollback to downloading, no need to retry here
    if (err != ERR_OK) {
        derror_f("app({}) partition({}) ingestion files failed, error = {}", app_name, pid, err);
        tasking::enqueue(
            LPC_META_STATE_NORMAL,
            _meta_svc->tracker(),
            std::bind(&bulk_load_service::try_rollback_to_downloading, this, app_name, pid));
        return;
    }

    if (resp.err == ERR_TRY_AGAIN && resp.rocksdb_error != 0) {
        derror_f("app({}) partition({}) ingestion files failed while empty write, rocksdb error = "
                 "{}, retry it later",
                 app_name,
                 pid,
                 resp.rocksdb_error);
        tasking::enqueue(LPC_BULK_LOAD_INGESTION,
                         _meta_svc->tracker(),
                         std::bind(&bulk_load_service::partition_ingestion, this, app_name, pid),
                         0,
                         std::chrono::milliseconds(10));
        return;
    }

    // some unexpected errors happened, such as write empty write failed but rocksdb_error is ok
    // stop bulk load process with failed
    if (resp.err != ERR_OK || resp.rocksdb_error != 0) {
        derror_f("app({}) partition({}) failed to ingestion files, error = {}, rocksdb error = {}",
                 app_name,
                 pid,
                 resp.err,
                 resp.rocksdb_error);
        tasking::enqueue(
            LPC_META_STATE_NORMAL,
            _meta_svc->tracker(),
            std::bind(&bulk_load_service::handle_bulk_load_failed, this, pid.get_app_id()));
        return;
    }

    ddebug_f("app({}) partition({}) receive ingestion response succeed", app_name, pid);
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::remove_bulk_load_dir_on_remote_storage(int32_t app_id,
                                                               const std::string &app_name)
{
    std::string bulk_load_path = get_app_bulk_load_path(app_id);
    _meta_svc->get_meta_storage()->delete_node_recursively(
        std::move(bulk_load_path), [this, app_id, app_name, bulk_load_path]() {
            ddebug_f("remove app({}) bulk load dir {} succeed", app_name, bulk_load_path);
            reset_local_bulk_load_states(app_id, app_name);
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::remove_bulk_load_dir_on_remote_storage(std::shared_ptr<app_state> app,
                                                               bool set_app_not_bulk_loading)
{
    std::string bulk_load_path = get_app_bulk_load_path(app->app_id);
    _meta_svc->get_meta_storage()->delete_node_recursively(
        std::move(bulk_load_path), [this, app, set_app_not_bulk_loading, bulk_load_path]() {
            ddebug_f("remove app({}) bulk load dir {} succeed", app->app_name, bulk_load_path);
            reset_local_bulk_load_states(app->app_id, app->app_name);
            if (set_app_not_bulk_loading) {
                update_app_not_bulk_loading_on_remote_storage(std::move(app));
            }
        });
}

// ThreadPool: THREAD_POOL_META_STATE
template <typename T>
inline void erase_map_elem_by_id(int32_t app_id, std::unordered_map<gpid, T> &mymap)
{
    for (auto iter = mymap.begin(); iter != mymap.end();) {
        if (iter->first.get_app_id() == app_id) {
            mymap.erase(iter++);
        } else {
            iter++;
        }
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::reset_local_bulk_load_states(int32_t app_id, const std::string &app_name)
{
    zauto_write_lock l(_lock);
    _app_bulk_load_info.erase(app_id);
    _apps_in_progress_count.erase(app_id);
    _apps_pending_sync_flag.erase(app_id);
    erase_map_elem_by_id(app_id, _partitions_pending_sync_flag);
    erase_map_elem_by_id(app_id, _partitions_bulk_load_state);
    erase_map_elem_by_id(app_id, _partition_bulk_load_info);
    erase_map_elem_by_id(app_id, _partitions_total_download_progress);
    erase_map_elem_by_id(app_id, _partitions_cleaned_up);
    _apps_rolling_back.erase(app_id);
    _apps_rollback_count.erase(app_id);
    _apps_ingesting_count.erase(app_id);
    _apps_cleaning_up.erase(app_id);
    _bulk_load_app_id.erase(app_id);
    ddebug_f("reset local app({}) bulk load context", app_name);
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::update_app_not_bulk_loading_on_remote_storage(
    std::shared_ptr<app_state> app)
{
    app_info info = *app;
    info.__set_is_bulk_loading(false);

    blob value = dsn::json::json_forwarder<app_info>::encode(info);
    _meta_svc->get_meta_storage()->set_data(
        _state->get_app_path(*app), std::move(value), [app, this]() {
            zauto_write_lock l(app_lock());
            app->is_bulk_loading = false;
            ddebug_f("app({}) update app is_bulk_loading to false", app->app_name);
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::on_control_bulk_load(control_bulk_load_rpc rpc)
{
    const std::string &app_name = rpc.request().app_name;
    const auto &control_type = rpc.request().type;
    auto &response = rpc.response();
    response.err = ERR_OK;

    std::shared_ptr<app_state> app = _state->get_app(app_name);
    if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
        derror_f("app({}) is not existed or not available", app_name);
        response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
        response.__set_hint_msg(fmt::format("app({}) is not existed or not available", app_name));
        return;
    }

    if (!app->is_bulk_loading) {
        derror_f("app({}) is not executing bulk load", app_name);
        response.err = ERR_INACTIVE_STATE;
        response.__set_hint_msg(fmt::format("app({}) is not executing bulk load", app_name));
        return;
    }
    int32_t app_id = app->app_id;

    zauto_write_lock l(_lock);
    const auto &app_status = get_app_bulk_load_status_unlocked(app_id);
    switch (control_type) {
    case bulk_load_control_type::BLC_PAUSE: {
        if (app_status != bulk_load_status::BLS_DOWNLOADING) {
            auto hint_msg = fmt::format("can not pause bulk load for app({}) with status({})",
                                        app_name,
                                        dsn::enum_to_string(app_status));
            derror_f("{}", hint_msg);
            response.err = ERR_INVALID_STATE;
            response.__set_hint_msg(hint_msg);
            return;
        }
        ddebug_f("app({}) start to pause bulk load", app_name);
        update_app_status_on_remote_storage_unlocked(app_id, bulk_load_status::BLS_PAUSING);
    } break;
    case bulk_load_control_type::BLC_RESTART: {
        if (app_status != bulk_load_status::BLS_PAUSED) {
            auto hint_msg = fmt::format("can not restart bulk load for app({}) with status({})",
                                        app_name,
                                        dsn::enum_to_string(app_status));
            derror_f("{}", hint_msg);
            response.err = ERR_INVALID_STATE;
            response.__set_hint_msg(hint_msg);
            return;
        }
        ddebug_f("app({}) restart bulk load", app_name);
        update_app_status_on_remote_storage_unlocked(
            app_id, bulk_load_status::BLS_DOWNLOADING, true);
    } break;
    case bulk_load_control_type::BLC_CANCEL:
        if (app_status != bulk_load_status::BLS_DOWNLOADING &&
            app_status != bulk_load_status::BLS_PAUSED) {
            auto hint_msg = fmt::format("can not cancel bulk load for app({}) with status({})",
                                        app_name,
                                        dsn::enum_to_string(app_status));
            derror_f("{}", hint_msg);
            response.err = ERR_INVALID_STATE;
            response.__set_hint_msg(hint_msg);
            return;
        }
    case bulk_load_control_type::BLC_FORCE_CANCEL: {
        ddebug_f("app({}) start to {} cancel bulk load, original status = {}",
                 app_name,
                 control_type == bulk_load_control_type::BLC_FORCE_CANCEL ? "force" : "",
                 dsn::enum_to_string(app_status));
        update_app_status_on_remote_storage_unlocked(
            app_id, bulk_load_status::BLS_CANCELED, app_status == bulk_load_status::BLS_PAUSED);
    } break;
    default:
        break;
    }
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
{
    const auto &request = rpc.request();
    const std::string &app_name = request.app_name;

    query_bulk_load_response &response = rpc.response();
    response.err = ERR_OK;
    response.app_name = app_name;

    int32_t app_id, partition_count, max_replica_count;
    {
        zauto_read_lock l(app_lock());
        std::shared_ptr<app_state> app = _state->get_app(app_name);

        if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
            auto hint_msg = fmt::format("app({}) is not existed or not available", app_name);
            derror_f("{}", hint_msg);
            response.err = app == nullptr ? ERR_APP_NOT_EXIST : ERR_APP_DROPPED;
            response.__set_hint_msg(hint_msg);
            return;
        }

        if (!app->is_bulk_loading) {
            auto hint_msg = fmt::format("app({}) is not during bulk load", app_name);
            derror_f("{}", hint_msg);
            response.err = ERR_INVALID_STATE;
            response.__set_hint_msg(hint_msg);
            return;
        }

        app_id = app->app_id;
        partition_count = app->partition_count;
        max_replica_count = app->max_replica_count;
    }

    zauto_read_lock l(_lock);
    response.max_replica_count = max_replica_count;
    response.app_status = get_app_bulk_load_status_unlocked(app_id);

    response.partitions_status.resize(partition_count);
    for (const auto &kv : _partition_bulk_load_info) {
        if (kv.first.get_app_id() == app_id) {
            response.partitions_status[kv.first.get_partition_index()] = kv.second.status;
        }
    }

    response.bulk_load_states.resize(partition_count);
    for (const auto &kv : _partitions_bulk_load_state) {
        if (kv.first.get_app_id() == app_id) {
            response.bulk_load_states[kv.first.get_partition_index()] = kv.second;
        }
    }

    ddebug_f("query app({}) bulk_load_status({}) succeed",
             app_name,
             dsn::enum_to_string(response.app_status));
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::create_bulk_load_root_dir()
{
    blob value = blob();
    std::string path = _bulk_load_root;
    _sync_bulk_load_storage->create_node(std::move(path), std::move(value), [this]() {
        ddebug_f("create bulk load root({}) succeed", _bulk_load_root);
        sync_apps_from_remote_storage();
    });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::sync_apps_from_remote_storage()
{
    std::string path = _bulk_load_root;
    _sync_bulk_load_storage->get_children(
        std::move(path), [this](bool flag, const std::vector<std::string> &children) {
            if (flag && children.size() > 0) {
                ddebug_f("There are {} apps need to sync bulk load status", children.size());
                for (const auto &elem : children) {
                    int32_t app_id = boost::lexical_cast<int32_t>(elem);
                    ddebug_f("start to sync app({}) bulk load status", app_id);
                    do_sync_app(app_id);
                }
            }
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::do_sync_app(int32_t app_id)
{
    std::string app_path = get_app_bulk_load_path(app_id);
    _sync_bulk_load_storage->get_data(std::move(app_path), [this, app_id](const blob &value) {
        app_bulk_load_info ainfo;
        dsn::json::json_forwarder<app_bulk_load_info>::decode(value, ainfo);
        {
            zauto_write_lock l(_lock);
            _bulk_load_app_id.insert(app_id);
            _app_bulk_load_info[app_id] = ainfo;
        }
        sync_partitions_from_remote_storage(ainfo.app_id, ainfo.app_name);
    });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::sync_partitions_from_remote_storage(int32_t app_id,
                                                            const std::string &app_name)
{
    std::string app_path = get_app_bulk_load_path(app_id);
    _sync_bulk_load_storage->get_children(
        std::move(app_path),
        [this, app_path, app_id, app_name](bool flag, const std::vector<std::string> &children) {
            ddebug_f("app(name={},app_id={}) has {} partition bulk load info to be synced",
                     app_name,
                     app_id,
                     children.size());
            for (const auto &child_pidx : children) {
                int32_t pidx = boost::lexical_cast<int32_t>(child_pidx);
                std::string partition_path = get_partition_bulk_load_path(app_path, pidx);
                do_sync_partition(gpid(app_id, pidx), partition_path);
            }
        });
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::do_sync_partition(const gpid &pid, std::string &partition_path)
{
    _sync_bulk_load_storage->get_data(std::move(partition_path), [this, pid](const blob &value) {
        partition_bulk_load_info pinfo;
        dsn::json::json_forwarder<partition_bulk_load_info>::decode(value, pinfo);
        {
            zauto_write_lock l(_lock);
            _partition_bulk_load_info[pid] = pinfo;
        }
    });
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::try_to_continue_bulk_load()
{
    FAIL_POINT_INJECT_F("meta_try_to_continue_bulk_load", [](dsn::string_view) {});
    for (const auto app_id : _bulk_load_app_id) {
        app_bulk_load_info ainfo = _app_bulk_load_info[app_id];
        // <partition_index, partition_bulk_load_info>
        std::unordered_map<int32_t, partition_bulk_load_info> pinfo_map;
        for (const auto &kv : _partition_bulk_load_info) {
            if (kv.first.get_app_id() == app_id) {
                pinfo_map[kv.first.get_partition_index()] = kv.second;
            }
        }
        try_to_continue_app_bulk_load(ainfo, pinfo_map);
    }
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::try_to_continue_app_bulk_load(
    const app_bulk_load_info &ainfo,
    const std::unordered_map<int32_t, partition_bulk_load_info> &pinfo_map)
{
    std::shared_ptr<app_state> app;
    {
        zauto_read_lock l(app_lock());
        app = _state->get_app(ainfo.app_name);
    }
    // if app is not available, remove bulk load dir
    if (app == nullptr || app->status != app_status::AS_AVAILABLE) {
        derror_f(
            "app(name={},app_id={}) is not existed or not available", ainfo.app_name, ainfo.app_id);
        if (app == nullptr) {
            remove_bulk_load_dir_on_remote_storage(ainfo.app_id, ainfo.app_name);
        } else {
            remove_bulk_load_dir_on_remote_storage(std::move(app), true);
        }
        return;
    }

    // check app bulk load info
    if (!validate_app(app->app_id, app->partition_count, ainfo, pinfo_map.size())) {
        remove_bulk_load_dir_on_remote_storage(std::move(app), true);
        return;
    }

    // index of the partition whose bulk load status is different from app's bulk load status
    std::unordered_set<int32_t> different_status_pidx_set;
    for (const auto &kv : pinfo_map) {
        if (kv.second.status != ainfo.status) {
            different_status_pidx_set.insert(kv.first);
        }
    }

    // check partition bulk load info
    if (!validate_partition(ainfo, pinfo_map, different_status_pidx_set.size())) {
        remove_bulk_load_dir_on_remote_storage(std::move(app), true);
        return;
    }

    tasking::enqueue(LPC_META_STATE_NORMAL,
                     _meta_svc->tracker(),
                     std::bind(&bulk_load_service::do_continue_app_bulk_load,
                               this,
                               ainfo,
                               pinfo_map,
                               different_status_pidx_set));
}

// ThreadPool: THREAD_POOL_META_SERVER
/*static*/ bool bulk_load_service::validate_app(int32_t app_id,
                                                int32_t partition_count,
                                                const app_bulk_load_info &ainfo,
                                                int32_t pinfo_count)
{
    // app id and partition from `app_bulk_load_info` is inconsistent with current app_info
    if (app_id != ainfo.app_id || partition_count != ainfo.partition_count) {
        derror_f("app({}) has different app_id or partition_count, bulk load app_id = {}, "
                 "partition_count = {}, current app_id = {}, partition_count = {}",
                 ainfo.app_name,
                 ainfo.app_id,
                 ainfo.partition_count,
                 app_id,
                 partition_count);
        return false;
    }

    // partition_bulk_load_info count should not be greater than partition_count
    if (partition_count < pinfo_count) {
        derror_f("app({}) has invalid count, app partition_count = {}, remote "
                 "partition_bulk_load_info count = {}",
                 ainfo.app_name,
                 partition_count,
                 pinfo_count);
        return false;
    }

    // partition_bulk_load_info count is not equal to partition_count can only be happended when app
    // status is downloading, consider the following condition:
    // when starting bulk load, meta server will create app_bulk_load_dir and
    // partition_bulk_load_dir on remote storage
    // however, meta server crash when create app directory and part of partition directory
    // when meta server recover, partition directory count is less than partition_count
    if (pinfo_count != partition_count && ainfo.status != bulk_load_status::BLS_DOWNLOADING) {
        derror_f("app({}) bulk_load_status = {}, but there are {} partitions lack "
                 "partition_bulk_load dir",
                 ainfo.app_name,
                 dsn::enum_to_string(ainfo.status),
                 partition_count - pinfo_count);
        return false;
    }

    return true;
}

// ThreadPool: THREAD_POOL_META_SERVER
/*static*/ bool bulk_load_service::validate_partition(
    const app_bulk_load_info &ainfo,
    const std::unordered_map<int32_t, partition_bulk_load_info> &pinfo_map,
    const int32_t different_status_count)
{
    const auto app_status = ainfo.status;
    bool is_valid = true;

    switch (app_status) {
    case bulk_load_status::BLS_DOWNLOADING:
        // if app status is downloading, partition status has no limit, because when bulk load meet
        // recoverable errors, will rollback to downloading
        // partition directory count is allowed less than partition_count, but it is impossible with
        // some partition bulk load status is not downloading and some partition directroy is
        // missing on remote storage
        if (ainfo.partition_count - pinfo_map.size() > 0 && different_status_count > 0) {
            derror_f("app({}) bulk_load_status = {}, there are {} partitions status is different "
                     "from app, and {} partitions not existed, this is invalid",
                     ainfo.app_name,
                     dsn::enum_to_string(app_status),
                     different_status_count,
                     ainfo.partition_count - pinfo_map.size());
            is_valid = false;
        }
        break;
    case bulk_load_status::BLS_DOWNLOADED:
    case bulk_load_status::BLS_INGESTING: {
        // if app status is downloaded, valid partition status is downloaded or ingesting
        // if app status is ingesting, valid partition status is ingesting or succeed
        const auto other_valid_status = (app_status == bulk_load_status::BLS_DOWNLOADED)
                                            ? bulk_load_status::BLS_INGESTING
                                            : bulk_load_status::BLS_SUCCEED;
        for (const auto &kv : pinfo_map) {
            if (kv.second.status != app_status && kv.second.status != other_valid_status) {
                derror_f("app({}) bulk_load_status = {}, but partition[{}] bulk_load_status = {}, "
                         "only {} and {} is valid",
                         ainfo.app_name,
                         app_status,
                         kv.first,
                         dsn::enum_to_string(kv.second.status),
                         dsn::enum_to_string(app_status),
                         dsn::enum_to_string(other_valid_status));
                is_valid = false;
                break;
            }
        }
    } break;
    case bulk_load_status::BLS_SUCCEED:
    case bulk_load_status::BLS_PAUSED:
        // if app status is succeed or paused, all partitions' status should not be different from
        // app's
        if (different_status_count > 0) {
            derror_f("app({}) bulk_load_status = {}, {} partitions status is different from app, "
                     "this is invalid",
                     ainfo.app_name,
                     dsn::enum_to_string(app_status),
                     different_status_count);
            is_valid = false;
        }
        break;
    default:
        // for other status, partition status has no limit
        break;
    }

    return is_valid;
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::do_continue_app_bulk_load(
    const app_bulk_load_info &ainfo,
    const std::unordered_map<int32_t, partition_bulk_load_info> &pinfo_map,
    const std::unordered_set<int32_t> &different_status_pidx_set)
{
    const int32_t app_id = ainfo.app_id;
    const int32_t partition_count = ainfo.partition_count;
    const auto app_status = ainfo.status;
    const int32_t different_count = different_status_pidx_set.size();
    const int32_t same_count = pinfo_map.size() - different_count;
    const int32_t invalid_count = partition_count - pinfo_map.size();

    ddebug_f(
        "app({}) continue bulk load, app_id = {}, partition_count = {}, status = {}, there are {} "
        "partitions have bulk_load_info, {} partitions have same status with app, {} "
        "partitions different",
        ainfo.app_name,
        app_id,
        partition_count,
        dsn::enum_to_string(app_status),
        pinfo_map.size(),
        same_count,
        different_count);

    // _apps_in_progress_count is used for updating app bulk load, when _apps_in_progress_count = 0
    // means app bulk load status can transfer to next stage, for example, when app status is
    // downloaded, and _apps_in_progress_count = 0, app status can turn to ingesting
    // see more in function `update_partition_status_on_remote_storage_reply`
    int32_t in_progress_partition_count = partition_count;
    if (app_status == bulk_load_status::BLS_DOWNLOADING) {
        if (invalid_count > 0) {
            // create missing partition, so the in_progress_count should be invalid_count
            in_progress_partition_count = invalid_count;
        } else if (different_count > 0) {
            // it is hard to distinguish that bulk load is normal downloading or rollback to
            // downloading before meta server crash, when app status is downloading, we consider
            // bulk load as rolling back to downloading for convenience, for partitions whose status
            // is not downloading, update them to downloading, so the in_progress_count should be
            // different_count
            in_progress_partition_count = different_count;
        }
    } else if (app_status == bulk_load_status::BLS_DOWNLOADED ||
               app_status == bulk_load_status::BLS_INGESTING ||
               app_status == bulk_load_status::BLS_SUCCEED) {
        // for app status is downloaded, when all partitions turn to ingesting, app partition will
        // turn to ingesting, so the in_progress_count should be same_count, ingesting and succeed
        // are same
        in_progress_partition_count = same_count;
    } // for other cases, in_progress_count should be partition_count
    {
        zauto_write_lock l(_lock);
        _apps_in_progress_count[app_id] = in_progress_partition_count;
        _apps_rollback_count[app_id] = 0;
        _apps_ingesting_count[app_id] = 0;
    }

    // if app is paused, no need to send bulk_load_request, just return
    if (app_status == bulk_load_status::BLS_PAUSED) {
        return;
    }

    // create all missing partitions then send request to all partitions
    if (app_status == bulk_load_status::BLS_DOWNLOADING && invalid_count > 0) {
        for (auto i = 0; i < partition_count; ++i) {
            if (pinfo_map.find(i) == pinfo_map.end()) {
                create_missing_partition_dir(ainfo.app_name, gpid(app_id, i), partition_count);
            }
        }
        return;
    }

    // update all partition status to app_status
    if ((app_status == bulk_load_status::BLS_FAILED ||
         app_status == bulk_load_status::BLS_CANCELED ||
         app_status == bulk_load_status::BLS_PAUSING ||
         app_status == bulk_load_status::BLS_DOWNLOADING) &&
        different_count > 0) {
        for (auto pidx : different_status_pidx_set) {
            update_partition_status_on_remote_storage(
                ainfo.app_name, gpid(app_id, pidx), app_status);
        }
    }

    // send bulk_load_request to all partitions
    for (auto i = 0; i < partition_count; ++i) {
        gpid pid = gpid(app_id, i);
        partition_bulk_load(ainfo.app_name, pid);
        if (app_status == bulk_load_status::BLS_INGESTING) {
            partition_ingestion(ainfo.app_name, pid);
        }
    }
}

// ThreadPool: THREAD_POOL_META_STATE
void bulk_load_service::create_missing_partition_dir(const std::string &app_name,
                                                     const gpid &pid,
                                                     int32_t partition_count)
{
    partition_bulk_load_info pinfo;
    pinfo.status = bulk_load_status::BLS_DOWNLOADING;
    blob value = dsn::json::json_forwarder<partition_bulk_load_info>::encode(pinfo);

    _meta_svc->get_meta_storage()->create_node(
        get_partition_bulk_load_path(pid),
        std::move(value),
        [app_name, pid, partition_count, pinfo, this]() {
            const int32_t app_id = pid.get_app_id();
            bool send_request = false;
            ddebug_f("app({}) create partition({}) bulk_load_info", app_name, pid);
            {
                zauto_write_lock l(_lock);
                _partition_bulk_load_info[pid] = pinfo;

                if (--_apps_in_progress_count[app_id] == 0) {
                    _apps_in_progress_count[app_id] = partition_count;
                    send_request = true;
                }
            }
            if (send_request) {
                ddebug_f("app({}) start to bulk load", app_name);
                for (auto i = 0; i < partition_count; ++i) {
                    partition_bulk_load(app_name, gpid(app_id, i));
                }
            }
        });
}

// ThreadPool: THREAD_POOL_META_SERVER
void bulk_load_service::check_app_bulk_load_states(std::shared_ptr<app_state> app,
                                                   bool is_app_bulk_loading)
{
    std::string app_path = get_app_bulk_load_path(app->app_id);
    _meta_svc->get_remote_storage()->node_exist(
        app_path, LPC_META_CALLBACK, [this, app_path, app, is_app_bulk_loading](error_code err) {
            if (err != ERR_OK && err != ERR_OBJECT_NOT_FOUND) {
                dwarn_f("check app({}) bulk load dir({}) failed, error = {}, try later",
                        app->app_name,
                        app_path,
                        err);
                tasking::enqueue(LPC_META_CALLBACK,
                                 nullptr,
                                 std::bind(&bulk_load_service::check_app_bulk_load_states,
                                           this,
                                           app,
                                           is_app_bulk_loading),
                                 0,
                                 std::chrono::seconds(1));
                return;
            }

            if (err == ERR_OBJECT_NOT_FOUND && is_app_bulk_loading) {
                derror_f("app({}): bulk load dir({}) not exist, but is_bulk_loading = {}, reset "
                         "app is_bulk_loading flag",
                         app->app_name,
                         app_path,
                         is_app_bulk_loading);
                update_app_not_bulk_loading_on_remote_storage(std::move(app));
                return;
            }

            if (err == ERR_OK && !is_app_bulk_loading) {
                derror_f("app({}): bulk load dir({}) exist, but is_bulk_loading = {}, remove "
                         "useless bulk load dir",
                         app->app_name,
                         app_path,
                         is_app_bulk_loading);
                remove_bulk_load_dir_on_remote_storage(std::move(app), false);
                return;
            }

            // Normal cases:
            // err = ERR_OBJECT_NOT_FOUND, is_app_bulk_load = false: app is not executing bulk load
            // err = ERR_OK, is_app_bulk_load = true: app used to be executing bulk load
        });
}

} // namespace replication
} // namespace dsn
