// 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 <brpc/controller.h>
#include <bvar/window.h>
#include <fmt/core.h>
#include <gen_cpp/cloud.pb.h>
#include <gen_cpp/olap_file.pb.h>
#include <google/protobuf/repeated_field.h>
#include <gtest/gtest.h>

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <random>
#include <string>
#include <thread>

#include "common/config.h"
#include "common/logging.h"
#include "common/util.h"
#include "cpp/sync_point.h"
#include "meta-service/meta_service.h"
#include "meta-service/meta_service_helper.h"
#include "meta-service/meta_service_txn.cpp"
#include "meta-store/keys.h"
#include "meta-store/mem_txn_kv.h"
#include "meta-store/txn_kv_error.h"
#include "mock_resource_manager.h"
#include "rate-limiter/rate_limiter.h"
#include "recycler/recycler.h"
#include "resource-manager/resource_manager.h"

using namespace doris::cloud;

namespace doris::cloud {
void repair_tablet_index(std::shared_ptr<TxnKv>& txn_kv, MetaServiceCode& code, std::string& msg,
                         const std::string& instance_id, int64_t db_id, int64_t txn_id,
                         const std::vector<int64_t>& tablet_ids, bool is_versioned_write);
};

static std::shared_ptr<TxnKv> txn_kv;
static doris::cloud::RecyclerThreadPoolGroup thread_group;

static std::shared_ptr<TxnKv> get_fdb_txn_kv() {
    if (txn_kv) {
        return txn_kv;
    }

    int ret = 0;
    doris::cloud::config::fdb_cluster_file_path = "fdb.cluster";
    auto fdb_txn_kv = std::dynamic_pointer_cast<doris::cloud::TxnKv>(
            std::make_shared<doris::cloud::FdbTxnKv>());
    if (fdb_txn_kv != nullptr) {
        ret = fdb_txn_kv->init();
        [&] { ASSERT_EQ(ret, 0); }();
    }
    [&] { ASSERT_NE(fdb_txn_kv.get(), nullptr); }();
    txn_kv = fdb_txn_kv;
    return fdb_txn_kv;
}

int main(int argc, char** argv) {
    const std::string conf_file = "doris_cloud.conf";
    if (!doris::cloud::config::init(conf_file.c_str(), true)) {
        std::cerr << "failed to init config file, conf=" << conf_file << std::endl;
        return -1;
    }

    config::enable_cloud_txn_lazy_commit = true;
    config::enable_txn_store_retry = false;

    config::label_keep_max_second = 0;
    config::force_immediate_recycle = true;
    config::log_immediate_flush = true;

    if (!doris::cloud::init_glog("txn_lazy_commit_test")) {
        std::cerr << "failed to init glog" << std::endl;
        return -1;
    }
    ::testing::InitGoogleTest(&argc, argv);

    // Initialize FDB
    get_fdb_txn_kv();

    auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    s3_producer_pool->start();
    auto recycle_tablet_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    recycle_tablet_pool->start();
    auto group_recycle_function_pool =
            std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism);
    group_recycle_function_pool->start();
    thread_group =
            RecyclerThreadPoolGroup(std::move(s3_producer_pool), std::move(recycle_tablet_pool),
                                    std::move(group_recycle_function_pool));

    int ret = RUN_ALL_TESTS();
    thread_group = RecyclerThreadPoolGroup();
    txn_kv.reset();
    return ret;
}
namespace doris::cloud {

std::unique_ptr<MetaServiceProxy> get_meta_service(std::shared_ptr<TxnKv> txn_kv,
                                                   bool mock_resource_mgr) {
    std::unique_ptr<Transaction> txn;
    EXPECT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    txn->remove("\x00", "\xfe"); // This is dangerous if the fdb is not correctly set
    EXPECT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    auto rs = mock_resource_mgr ? std::make_shared<MockResourceManager>(txn_kv)
                                : std::make_shared<ResourceManager>(txn_kv);
    auto rl = std::make_shared<RateLimiter>();
    auto snapshot = std::make_shared<SnapshotManager>(txn_kv);
    auto meta_service = std::make_unique<MetaServiceImpl>(txn_kv, rs, rl, snapshot);
    return std::make_unique<MetaServiceProxy>(std::move(meta_service));
}

static std::string next_rowset_id() {
    static int cnt = 0;
    return std::to_string(++cnt);
}

static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
                       int64_t partition_id, int64_t tablet_id) {
    auto tablet = req.add_tablet_metas();
    tablet->set_table_id(table_id);
    tablet->set_index_id(index_id);
    tablet->set_partition_id(partition_id);
    tablet->set_tablet_id(tablet_id);
    auto schema = tablet->mutable_schema();
    schema->set_schema_version(0);
    auto first_rowset = tablet->add_rs_metas();
    first_rowset->set_rowset_id(0); // required
    first_rowset->set_rowset_id_v2(next_rowset_id());
    first_rowset->set_start_version(0);
    first_rowset->set_end_version(1);
    first_rowset->mutable_tablet_schema()->CopyFrom(*schema);
}

static void create_tablet_without_db_id(MetaServiceProxy* meta_service, int64_t table_id,
                                        int64_t index_id, int64_t partition_id, int64_t tablet_id) {
    brpc::Controller cntl;
    CreateTabletsRequest req;
    CreateTabletsResponse res;
    add_tablet(req, table_id, index_id, partition_id, tablet_id);
    meta_service->create_tablets(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}

static void create_tablet_with_db_id(MetaServiceProxy* meta_service, int64_t db_id,
                                     int64_t table_id, int64_t index_id, int64_t partition_id,
                                     int64_t tablet_id) {
    brpc::Controller cntl;
    CreateTabletsRequest req;
    CreateTabletsResponse res;
    req.set_db_id(db_id);
    add_tablet(req, table_id, index_id, partition_id, tablet_id);
    meta_service->create_tablets(&cntl, &req, &res, nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
}

static doris::RowsetMetaCloudPB create_rowset(int64_t txn_id, int64_t tablet_id, int64_t index_id,
                                              int64_t partition_id, int64_t version = -1,
                                              int num_rows = 100) {
    doris::RowsetMetaCloudPB rowset;
    rowset.set_rowset_id(0); // required
    rowset.set_rowset_id_v2(next_rowset_id());
    rowset.set_tablet_id(tablet_id);
    rowset.set_partition_id(partition_id);
    rowset.set_index_id(index_id);
    rowset.set_txn_id(txn_id);
    if (version > 0) {
        rowset.set_start_version(version);
        rowset.set_end_version(version);
    }
    rowset.set_num_segments(0);
    rowset.set_num_rows(0);
    rowset.set_data_disk_size(0);
    rowset.mutable_tablet_schema()->set_schema_version(0);
    rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
    return rowset;
}

static doris::RowsetMetaCloudPB create_huge_rowset(int64_t txn_id, int64_t tablet_id,
                                                   int64_t index_id, int64_t partition_id,
                                                   int64_t version = -1, int num_rows = 100) {
    doris::RowsetMetaCloudPB rowset;
    rowset.set_rowset_id(0); // required
    rowset.set_rowset_id_v2(next_rowset_id());
    rowset.set_tablet_id(tablet_id);
    rowset.set_partition_id(partition_id);
    rowset.set_index_id(index_id);
    rowset.set_txn_id(txn_id);
    if (version > 0) {
        rowset.set_start_version(version);
        rowset.set_end_version(version);
    }
    rowset.set_num_segments(600);
    for (int i = 0; i < 600; i++) {
        auto ptr = rowset.add_segments_key_bounds();
        ptr->set_min_key("xxsqewqeqweeqwewqeqeq");
        ptr->set_max_key("dase23452rr234ewdw534523");
    }
    rowset.set_num_rows(0);
    rowset.set_data_disk_size(0);
    rowset.mutable_tablet_schema()->set_schema_version(0);
    rowset.set_txn_expiration(::time(nullptr)); // Required by DCHECK
    return rowset;
}

static void prepare_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
                           CreateRowsetResponse& res) {
    brpc::Controller cntl;
    auto arena = res.GetArena();
    auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
    req->mutable_rowset_meta()->CopyFrom(rowset);
    meta_service->prepare_rowset(&cntl, req, &res, nullptr);
    if (!arena) {
        delete req;
    }
}

static void commit_rowset(MetaServiceProxy* meta_service, const doris::RowsetMetaCloudPB& rowset,
                          CreateRowsetResponse& res) {
    brpc::Controller cntl;
    auto arena = res.GetArena();
    auto req = google::protobuf::Arena::CreateMessage<CreateRowsetRequest>(arena);
    req->mutable_rowset_meta()->CopyFrom(rowset);
    meta_service->commit_rowset(&cntl, req, &res, nullptr);
    if (!arena) {
        delete req;
    }
}

static std::shared_ptr<TxnKv> get_mem_txn_kv() {
    int ret = 0;
    // MemKv
    auto txn_kv = std::dynamic_pointer_cast<TxnKv>(std::make_shared<MemTxnKv>());
    if (txn_kv != nullptr) {
        ret = txn_kv->init();
        [&] { ASSERT_EQ(ret, 0); }();
    }
    [&] { ASSERT_NE(txn_kv.get(), nullptr); }();
    return txn_kv;
}

static void check_tablet_idx_db_id(std::unique_ptr<Transaction>& txn, int64_t db_id,
                                   int64_t tablet_id) {
    std::string mock_instance = "test_instance";
    std::string key = meta_tablet_idx_key({mock_instance, tablet_id});
    std::string val;
    ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
    TabletIndexPB tablet_idx_pb;
    tablet_idx_pb.ParseFromString(val);
    ASSERT_EQ(tablet_idx_pb.db_id(), db_id);
}

static void check_tablet_idx_without_db_id(std::unique_ptr<Transaction>& txn, int64_t tablet_id) {
    std::string mock_instance = "test_instance";
    std::string key = meta_tablet_idx_key({mock_instance, tablet_id});
    std::string val;
    ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
    TabletIndexPB tablet_idx_pb;
    tablet_idx_pb.ParseFromString(val);
    ASSERT_FALSE(tablet_idx_pb.has_db_id());
}

static void check_tmp_rowset_exist(std::unique_ptr<Transaction>& txn, int64_t tablet_id,
                                   int64_t txn_id) {
    std::string mock_instance = "test_instance";
    std::string key = meta_rowset_tmp_key({mock_instance, txn_id, tablet_id});
    std::string val;
    ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
}

static void check_tmp_rowset_not_exist(std::unique_ptr<Transaction>& txn, int64_t tablet_id,
                                       int64_t txn_id) {
    std::string mock_instance = "test_instance";
    std::string key = meta_rowset_tmp_key({mock_instance, txn_id, tablet_id});
    std::string val;
    ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_KEY_NOT_FOUND);
}

static void check_rowset_meta_exist(std::unique_ptr<Transaction>& txn, int64_t tablet_id,
                                    int64_t end_version) {
    std::string mock_instance = "test_instance";
    std::string rowset_key = meta_rowset_key({mock_instance, tablet_id, end_version});
    std::string rowset_val;
    ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
}

static void check_rowset_meta_not_exist(std::unique_ptr<Transaction>& txn, int64_t tablet_id,
                                        int64_t end_version) {
    std::string mock_instance = "test_instance";
    std::string rowset_key = meta_rowset_key({mock_instance, tablet_id, end_version});
    std::string rowset_val;
    ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_KEY_NOT_FOUND);
}

[[maybe_unused]] static void check_txn_visible(std::unique_ptr<Transaction>& txn, int64_t db_id,
                                               int64_t txn_id, std::string label) {
    std::string mock_instance = "test_instance";
    std::string info_key, info_val;
    TxnInfoKeyInfo txn_info_key_info {mock_instance, db_id, txn_id};
    txn_info_key(txn_info_key_info, &info_key);
    ASSERT_EQ(txn->get(info_key, &info_val), TxnErrorCode::TXN_OK);
    TxnInfoPB txn_info_pb;
    txn_info_pb.ParseFromString(info_val);
    ASSERT_EQ(txn_info_pb.status(), TxnStatusPB::TXN_STATUS_VISIBLE);

    std::string label_key, label_val;
    txn_label_key({mock_instance, db_id, label}, &label_key);
    ASSERT_EQ(txn->get(label_key, &label_val), TxnErrorCode::TXN_OK);

    std::string index_key, index_val;
    index_key = txn_index_key({mock_instance, txn_id});
    ASSERT_EQ(txn->get(index_key, &index_val), TxnErrorCode::TXN_OK);

    std::string running_key, running_value;
    TxnRunningKeyInfo running_key_info {mock_instance, db_id, txn_id};
    txn_running_key(running_key_info, &running_key);
    ASSERT_EQ(txn->get(running_key, &running_value), TxnErrorCode::TXN_KEY_NOT_FOUND);

    std::string rec_txn_key, rec_txn_val;
    RecycleTxnKeyInfo recycle_txn_key_info {mock_instance, db_id, txn_id};
    recycle_txn_key(recycle_txn_key_info, &rec_txn_key);
    ASSERT_EQ(txn->get(rec_txn_key, &rec_txn_val), TxnErrorCode::TXN_OK);
}

[[maybe_unused]] static void check_txn_committed(std::unique_ptr<Transaction>& txn, int64_t db_id,
                                                 int64_t txn_id, std::string label) {
    std::string mock_instance = "test_instance";
    std::string info_key, info_val;
    TxnInfoKeyInfo txn_info_key_info {mock_instance, db_id, txn_id};
    txn_info_key(txn_info_key_info, &info_key);
    ASSERT_EQ(txn->get(info_key, &info_val), TxnErrorCode::TXN_OK);
    TxnInfoPB txn_info_pb;
    txn_info_pb.ParseFromString(info_val);
    ASSERT_EQ(txn_info_pb.status(), TxnStatusPB::TXN_STATUS_COMMITTED);

    std::string label_key, label_val;
    txn_label_key({mock_instance, db_id, label}, &label_key);
    ASSERT_EQ(txn->get(label_key, &label_val), TxnErrorCode::TXN_OK);

    std::string index_key, index_val;
    index_key = txn_index_key({mock_instance, txn_id});
    ASSERT_EQ(txn->get(index_key, &index_val), TxnErrorCode::TXN_OK);

    std::string running_key, running_value;
    TxnRunningKeyInfo running_key_info {mock_instance, db_id, txn_id};
    txn_running_key(running_key_info, &running_key);
    ASSERT_EQ(txn->get(running_key, &running_value), TxnErrorCode::TXN_OK);

    std::string rec_txn_key, rec_txn_val;
    RecycleTxnKeyInfo recycle_txn_key_info {mock_instance, db_id, txn_id};
    recycle_txn_key(recycle_txn_key_info, &rec_txn_key);
    ASSERT_EQ(txn->get(rec_txn_key, &rec_txn_val), TxnErrorCode::TXN_KEY_NOT_FOUND);
}

static void check_txn_not_exist(std::unique_ptr<Transaction>& txn, int64_t db_id, int64_t txn_id,
                                std::string label) {
    std::string mock_instance = "test_instance";

    std::string info_key, info_val;
    TxnInfoKeyInfo txn_info_key_info {mock_instance, db_id, txn_id};
    txn_info_key(txn_info_key_info, &info_key);
    ASSERT_EQ(txn->get(info_key, &info_val), TxnErrorCode::TXN_KEY_NOT_FOUND);

    std::string label_key, label_val;
    txn_label_key({mock_instance, db_id, label}, &label_key);
    ASSERT_EQ(txn->get(label_key, &label_val), TxnErrorCode::TXN_KEY_NOT_FOUND);

    std::string index_key, index_val;
    index_key = txn_index_key({mock_instance, txn_id});
    ASSERT_EQ(txn->get(index_key, &index_val), TxnErrorCode::TXN_KEY_NOT_FOUND);

    std::string running_key, running_value;
    TxnRunningKeyInfo running_key_info {mock_instance, db_id, txn_id};
    txn_running_key(running_key_info, &running_key);
    ASSERT_EQ(txn->get(running_key, &running_value), TxnErrorCode::TXN_KEY_NOT_FOUND);

    std::string rec_txn_key, rec_txn_val;
    RecycleTxnKeyInfo recycle_txn_key_info {mock_instance, db_id, txn_id};
    recycle_txn_key(recycle_txn_key_info, &rec_txn_key);
    ASSERT_EQ(txn->get(rec_txn_key, &rec_txn_val), TxnErrorCode::TXN_KEY_NOT_FOUND);
}

// Create a MULTI_VERSION_READ_WRITE instance and refresh the resource manager.
static void create_and_refresh_instance(
        MetaServiceProxy* service, std::string instance_id,
        MultiVersionStatus multi_version_status = MULTI_VERSION_READ_WRITE) {
    // write instance
    InstanceInfoPB instance_info;
    instance_info.set_instance_id(instance_id);
    instance_info.set_multi_version_status(multi_version_status);
    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
    txn->put(instance_key(instance_id), instance_info.SerializeAsString());
    ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);

    service->resource_mgr()->refresh_instance(instance_id);
    ASSERT_TRUE(service->resource_mgr()->is_version_write_enabled(instance_id));
}

static void get_table_version(MetaServiceProxy* meta_service, int64_t db_id, int64_t table_id,
                              int64_t& version) {
    brpc::Controller ctrl;
    GetVersionRequest req;
    req.set_db_id(db_id);
    req.set_table_id(table_id);
    req.set_is_table_version(true);
    GetVersionResponse resp;
    meta_service->get_version(&ctrl, &req, &resp, nullptr);
    ASSERT_EQ(resp.status().code(), MetaServiceCode::OK)
            << ", get table version res=" << resp.ShortDebugString();
    version = resp.version();
}

TEST(TxnLazyCommitTest, CreateTabletWithDbIdTest) {
    auto txn_kv = get_mem_txn_kv();
    auto meta_service = get_meta_service(txn_kv, true);
    int db_id = 1000313313;
    int table_id = 1001414121;
    int index_id = 1002316473;
    int partition_id = 10035151;

    // mock rowset and tablet
    int64_t tablet_id_base = 11414703;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (int i = 0; i < 2001; ++i) {
        int64_t tablet_id = tablet_id_base + i;
        check_tablet_idx_db_id(txn, db_id, tablet_id);
    }
}

TEST(TxnLazyCommitTest, CreateTabletWithoutDbIdTest) {
    auto txn_kv = get_mem_txn_kv();
    auto meta_service = get_meta_service(txn_kv, true);
    int table_id = 3131;
    int index_id = 4131;
    int partition_id = 3131;

    // mock rowset and tablet
    int64_t tablet_id_base = 42411890;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
    }

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (int i = 0; i < 2001; ++i) {
        int64_t tablet_id = tablet_id_base + i;
        check_tablet_idx_without_db_id(txn, tablet_id);
    }
}

TEST(TxnLazyCommitTest, RepairTabletIndexTest) {
    auto txn_kv = get_mem_txn_kv();
    auto meta_service = get_meta_service(txn_kv, true);
    int64_t db_id = 21318977;
    int64_t table_id = 46714;
    int64_t index_id = 83645;
    int64_t partition_id = 123131;
    std::string label = "test_repair_tablet_index";
    int64_t txn_id = 0;
    std::string mock_instance = "test_instance";

    {
        brpc::Controller cntl;
        BeginTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");

        TxnInfoPB txn_info;
        txn_info.set_db_id(db_id);
        txn_info.set_label(label);
        txn_info.add_table_ids(table_id);
        txn_info.set_timeout_ms(6000);
        req.mutable_txn_info()->CopyFrom(txn_info);
        BeginTxnResponse res;
        meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        txn_id = res.txn_id();
    }

    ASSERT_GT(txn_id, 0);

    // mock rowset and tablet
    int64_t tablet_id_base = 87134121;
    std::vector<std::pair<std::string, doris::RowsetMetaCloudPB>> tmp_rowsets_meta;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        tmp_rowsets_meta.push_back(std::make_pair("mock_tmp_rowset_key", tmp_rowset));
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_without_db_id(txn, tablet_id);
        }
    }

    MetaServiceCode code = MetaServiceCode::UNDEFINED_ERR;
    std::string msg;
    std::vector<int64_t> tablet_ids;
    for (int i = 0; i < 2001; ++i) {
        tablet_ids.push_back(tablet_id_base + i);
    }
    repair_tablet_index(txn_kv, code, msg, mock_instance, db_id, txn_id, tablet_ids, false);
    ASSERT_EQ(code, MetaServiceCode::OK);

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithoutDbIdTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 3131397513;
    int64_t table_id = 3213867;
    int64_t index_id = 123513;
    int64_t partition_id = 113123;
    bool commit_txn_eventually_finish_hit = false;
    bool last_pending_txn_id_hit = false;
    int repair_tablet_idx_count = 0;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        LOG(INFO) << "need_repair_tablet_idx:" << need_repair_tablet_idx;
        if (repair_tablet_idx_count == 0) {
            ASSERT_TRUE(need_repair_tablet_idx);
            repair_tablet_idx_count++;
        } else {
            ASSERT_FALSE(need_repair_tablet_idx);
        }
    });

    sp->set_call_back("commit_txn_eventually::last_pending_txn_id", [&](auto&& args) {
        int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
        ASSERT_EQ(last_pending_txn_id, 0);
        last_pending_txn_id_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_commit_txn_eventually");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 1103;
    for (int i = 0; i < 2999; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
        }
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        for (int i = 0; i < 2999; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            req.add_base_tablet_ids(tablet_id);
        }
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_GE(repair_tablet_idx_count, 0);
        ASSERT_TRUE(last_pending_txn_id_hit);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);

        ASSERT_EQ(res.table_stats().size(), 1);
        int64_t table_version = 0;
        get_table_version(meta_service.get(), db_id, table_id, table_version);
        ASSERT_EQ(table_version, 1);
        ASSERT_EQ(res.table_stats()[0].table_version(), table_version);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 2999; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithAbortedTest) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 55432134;
    int64_t table_id = 326843;
    int64_t index_id = 34345678;
    int64_t partition_id = 212343;

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_commit_txn_eventually");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 372323;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
        }
    }

    {
        brpc::Controller cntl;
        AbortTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_txn_id(txn_id);
        req.set_reason("test");
        AbortTxnResponse res;
        meta_service->abort_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_EQ(res.txn_info().status(), TxnStatusPB::TXN_STATUS_ABORTED);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            req.add_base_tablet_ids(tablet_id);
        }
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::TXN_ALREADY_ABORTED);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
            check_rowset_meta_not_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithDbIdTest) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 7651485414;
    int64_t table_id = 31478952181;
    int64_t index_id = 89894141;
    int64_t partition_id = 1241241;
    bool commit_txn_eventually_finish_hit = false;
    bool last_pending_txn_id_hit = false;
    int repair_tablet_idx_count = 0;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        ASSERT_FALSE(need_repair_tablet_idx);
        repair_tablet_idx_count++;
    });

    sp->set_call_back("commit_txn_eventually::last_pending_txn_id", [&](auto&& args) {
        int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
        ASSERT_EQ(last_pending_txn_id, 0);
        last_pending_txn_id_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_commit_txn_eventually2");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 3131124;
    for (int i = 0; i < 2048; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_GE(repair_tablet_idx_count, 0);
        ASSERT_TRUE(last_pending_txn_id_hit);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 2048; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitVersionedReadTest, CommitTxnEventually) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 7651485414;
    int64_t table_id = 31478952181;
    int64_t index_id = 89894141;
    int64_t partition_id = 1241241;
    bool commit_txn_eventually_finish_hit = false;
    bool last_pending_txn_id_hit = false;
    int repair_tablet_idx_count = 0;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        ASSERT_FALSE(need_repair_tablet_idx);
        repair_tablet_idx_count++;
    });

    sp->set_call_back("commit_txn_eventually::last_pending_txn_id", [&](auto&& args) {
        int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
        ASSERT_EQ(last_pending_txn_id, 0);
        last_pending_txn_id_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, false);
    std::string instance_id = "test_instance";
    std::string cloud_unique_id = "1:test_instance:1";
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    SyncPoint::get_instance()->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = instance_id;
        ret->second = true;
    });
    SyncPoint::get_instance()->enable_processing();
    create_and_refresh_instance(meta_service.get(), instance_id);

    int txn_id = 0;
    {
        // Begin transaction
        brpc::Controller cntl;
        BeginTxnRequest req;
        req.set_cloud_unique_id(cloud_unique_id);
        TxnInfoPB txn_info_pb;
        txn_info_pb.set_db_id(db_id);
        txn_info_pb.set_label("test_label_commit_txn_eventually2");
        txn_info_pb.add_table_ids(table_id);
        txn_info_pb.set_timeout_ms(36000);
        req.mutable_txn_info()->CopyFrom(txn_info_pb);
        BeginTxnResponse res;
        meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        txn_id = res.txn_id();
    }

    // mock rowset and tablet
    int64_t tablet_id_base = 3131124;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_GE(repair_tablet_idx_count, 0);
        ASSERT_TRUE(last_pending_txn_id_hit);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);

        ASSERT_EQ(res.table_stats().size(), 1);
        int64_t table_version = 0;
        get_table_version(meta_service.get(), db_id, table_id, table_version);
        ASSERT_EQ(res.table_stats()[0].table_version(), table_version);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);

            // Check versioned rowset meta exists.
            std::string rowset_key = versioned::meta_rowset_load_key({mock_instance, tablet_id, 2});
            doris::RowsetMetaCloudPB rowset_val;
            Versionstamp versionstamp;
            ASSERT_EQ(versioned::document_get(txn.get(), rowset_key, &rowset_val, &versionstamp),
                      TxnErrorCode::TXN_OK);
        }
    }

    {
        // Get the partition version
        brpc::Controller ctrl;
        GetVersionRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(1);
        req.set_table_id(table_id);
        req.set_partition_id(partition_id);

        GetVersionResponse resp;
        meta_service->get_version(&ctrl, &req, &resp, nullptr);

        ASSERT_EQ(resp.status().code(), MetaServiceCode::OK)
                << " status is " << resp.status().DebugString();
        ASSERT_GT(resp.version(), 1);
    }
}

TEST(TxnLazyCommitVersionedReadTest, DISABLED_CommitTxnEventuallyWithoutDbIdTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 3131397513;
    int64_t table_id = 3213867;
    int64_t index_id = 123513;
    int64_t partition_id = 113123;
    bool commit_txn_eventually_finish_hit = false;
    bool last_pending_txn_id_hit = false;
    int repair_tablet_idx_count = 0;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        if (repair_tablet_idx_count == 0) {
            ASSERT_TRUE(need_repair_tablet_idx);
            repair_tablet_idx_count++;
        } else {
            ASSERT_FALSE(need_repair_tablet_idx);
        }
    });

    sp->set_call_back("commit_txn_eventually::last_pending_txn_id", [&](auto&& args) {
        int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
        ASSERT_EQ(last_pending_txn_id, 0);
        last_pending_txn_id_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, false);
    std::string instance_id = "test_instance";
    std::string cloud_unique_id = "1:test_instance:1";
    DORIS_CLOUD_DEFER {
        SyncPoint::get_instance()->clear_all_call_backs();
    };
    SyncPoint::get_instance()->set_call_back("get_instance_id", [&](auto&& args) {
        auto* ret = try_any_cast_ret<std::string>(args);
        ret->first = instance_id;
        ret->second = true;
    });
    SyncPoint::get_instance()->enable_processing();
    create_and_refresh_instance(meta_service.get(), instance_id, MULTI_VERSION_WRITE_ONLY);

    int64_t txn_id = 0;
    {
        brpc::Controller cntl;
        BeginTxnRequest req;
        req.set_cloud_unique_id(cloud_unique_id);
        TxnInfoPB txn_info_pb;
        txn_info_pb.set_db_id(db_id);
        txn_info_pb.set_label("test_label_commit_txn_eventually");
        txn_info_pb.add_table_ids(table_id);
        txn_info_pb.set_timeout_ms(36000);
        req.mutable_txn_info()->CopyFrom(txn_info_pb);
        BeginTxnResponse res;
        meta_service->begin_txn(&cntl, &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        txn_id = res.txn_id();
    }

    // mock rowset and tablet
    int64_t tablet_id_base = 1103;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
        }
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id(cloud_unique_id);
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            req.add_base_tablet_ids(tablet_id);
        }
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_GE(repair_tablet_idx_count, 0);
        ASSERT_TRUE(last_pending_txn_id_hit);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);

            {
                std::string mock_instance = "test_instance";
                std::string key = versioned::tablet_index_key({mock_instance, tablet_id});
                std::string val;
                ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
                TabletIndexPB tablet_idx_pb;
                tablet_idx_pb.ParseFromString(val);
                ASSERT_EQ(tablet_idx_pb.db_id(), db_id);
            }
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnImmediatelyRepairTabletIndexTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 983153141;
    int64_t table_id = 71419093;
    int64_t index_id = 80124;
    int64_t partition_id = 8989313;
    bool commit_txn_immediatelly_hit = false;
    int repair_tablet_idx_count = 0;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_immediately::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        if (repair_tablet_idx_count == 0) {
            ASSERT_TRUE(need_repair_tablet_idx);
        } else {
            ASSERT_FALSE(need_repair_tablet_idx);
        }
        repair_tablet_idx_count++;
    });
    sp->set_call_back("commit_txn_immediately::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_immediatelly_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_commit_txn_immediatelly");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 31311414;
    for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(commit_txn_immediatelly_hit);
        ASSERT_EQ(repair_tablet_idx_count, 2);
        ASSERT_TRUE(res.has_is_lazy_commit());
        ASSERT_FALSE(res.is_lazy_commit());
        ASSERT_FALSE(res.has_is_lazy_commit_incomplete());
        ASSERT_FALSE(res.is_lazy_commit_incomplete());
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnImmediatelyRepairOnlyMissingDbIdTest) {
    auto txn_kv = get_mem_txn_kv();
    auto meta_service = get_meta_service(txn_kv, true);
    const int64_t db_id = 983153143;
    const int64_t table_id = 71419095;
    const int64_t index_id = 80126;
    const int64_t partition_id = 8989316;
    const int64_t tablet_id_base = 31311420;
    const int64_t bad_tablet_id = tablet_id_base + 1;
    const std::string instance_id = "test_instance";

    brpc::Controller cntl;
    BeginTxnRequest begin_req;
    begin_req.set_cloud_unique_id("test_cloud_unique_id");
    auto* txn_info = begin_req.mutable_txn_info();
    txn_info->set_db_id(db_id);
    txn_info->set_label("test_commit_txn_repair_only_missing_db_id");
    txn_info->add_table_ids(table_id);
    txn_info->set_timeout_ms(36000);
    BeginTxnResponse begin_res;
    meta_service->begin_txn(&cntl, &begin_req, &begin_res, nullptr);
    ASSERT_EQ(begin_res.status().code(), MetaServiceCode::OK);
    const int64_t txn_id = begin_res.txn_id();

    for (int i = 0; i < 3; ++i) {
        const int64_t tablet_id = tablet_id_base + i;
        if (tablet_id == bad_tablet_id) {
            create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                        tablet_id);
        } else {
            create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                     tablet_id);
        }
        auto rowset = create_rowset(txn_id, tablet_id, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    std::vector<std::string> original_index_values(3);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_tablet_idx_without_db_id(txn, bad_tablet_id);
        for (int i = 0; i < 3; ++i) {
            auto key = meta_tablet_idx_key({instance_id, tablet_id_base + i});
            ASSERT_EQ(txn->get(key, &original_index_values[i]), TxnErrorCode::TXN_OK);
        }
    }

    std::vector<bool> repair_required;
    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_immediately::need_repair_tablet_idx", [&](auto&& args) {
        repair_required.push_back(*try_any_cast<bool*>(args[0]));
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    const auto repair_count_before = g_bvar_ms_repair_tablet_index.get_value();
    CommitTxnRequest commit_req;
    commit_req.set_cloud_unique_id("test_cloud_unique_id");
    commit_req.set_db_id(db_id);
    commit_req.set_txn_id(txn_id);
    commit_req.set_is_2pc(false);
    commit_req.set_enable_txn_lazy_commit(false);
    CommitTxnResponse commit_res;
    meta_service->commit_txn(&cntl, &commit_req, &commit_res, nullptr);
    ASSERT_EQ(commit_res.status().code(), MetaServiceCode::OK);
    ASSERT_FALSE(commit_res.is_lazy_commit());
    ASSERT_EQ(repair_required, (std::vector<bool> {true, false}));
    ASSERT_EQ(g_bvar_ms_repair_tablet_index.get_value() - repair_count_before, 1);

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    for (int i = 0; i < 3; ++i) {
        const int64_t tablet_id = tablet_id_base + i;
        check_tablet_idx_db_id(txn, db_id, tablet_id);
        std::string value;
        ASSERT_EQ(txn->get(meta_tablet_idx_key({instance_id, tablet_id}), &value),
                  TxnErrorCode::TXN_OK);
        if (tablet_id == bad_tablet_id) {
            TabletIndexPB expected;
            ASSERT_TRUE(expected.ParseFromString(original_index_values[i]));
            expected.set_db_id(db_id);
            ASSERT_EQ(value, expected.SerializeAsString());
        } else {
            ASSERT_EQ(value, original_index_values[i]);
        }
        check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
        check_rowset_meta_exist(txn, tablet_id, 2);
    }
}

TEST(TxnLazyCommitTest, CommitTxnWithSubTxnRepairTabletIndexTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 983153142;
    int64_t table_id = 71419094;
    int64_t index_id = 80125;
    int64_t partition_id1 = 8989314;
    int64_t partition_id2 = 8989315;
    int64_t tablet_id1 = 31311415;
    int64_t tablet_id2 = 31311416;
    int repair_tablet_idx_count = 0;
    bool commit_txn_with_sub_txn_hit = false;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_with_sub_txn::need_repair_tablet_idx", [&](auto&& args) {
        bool need_repair_tablet_idx = *try_any_cast<bool*>(args[0]);
        if (repair_tablet_idx_count == 0) {
            ASSERT_TRUE(need_repair_tablet_idx);
        } else {
            ASSERT_FALSE(need_repair_tablet_idx);
        }
        repair_tablet_idx_count++;
    });
    sp->set_call_back("commit_txn_with_sub_txn::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_with_sub_txn_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    int64_t txn_id = 0;
    {
        brpc::Controller cntl;
        BeginTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        TxnInfoPB txn_info_pb;
        txn_info_pb.set_db_id(db_id);
        txn_info_pb.set_label("test_commit_txn_with_sub_txn_repair");
        txn_info_pb.add_table_ids(table_id);
        txn_info_pb.set_timeout_ms(36000);
        req.mutable_txn_info()->CopyFrom(txn_info_pb);
        BeginTxnResponse res;
        meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        txn_id = res.txn_id();
    }

    create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id1, tablet_id1);
    auto tmp_rowset1 = create_rowset(txn_id, tablet_id1, index_id, partition_id1);
    CreateRowsetResponse rowset_res;
    prepare_rowset(meta_service.get(), tmp_rowset1, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
    commit_rowset(meta_service.get(), tmp_rowset1, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);

    int64_t sub_txn_id = 0;
    {
        brpc::Controller cntl;
        BeginSubTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_txn_id(txn_id);
        req.set_sub_txn_num(0);
        req.set_db_id(db_id);
        req.set_label("test_commit_txn_with_sub_txn_repair_sub");
        req.mutable_table_ids()->Add(table_id);
        BeginSubTxnResponse res;
        meta_service->begin_sub_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(res.has_sub_txn_id());
        sub_txn_id = res.sub_txn_id();
    }

    create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id2, tablet_id2);
    auto tmp_rowset2 = create_rowset(sub_txn_id, tablet_id2, index_id, partition_id2);
    rowset_res.Clear();
    prepare_rowset(meta_service.get(), tmp_rowset2, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
    commit_rowset(meta_service.get(), tmp_rowset2, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_txn_load(true);

        SubTxnInfo sub_txn_info1;
        sub_txn_info1.set_sub_txn_id(txn_id);
        sub_txn_info1.set_table_id(table_id);
        sub_txn_info1.mutable_base_tablet_ids()->Add(tablet_id1);

        SubTxnInfo sub_txn_info2;
        sub_txn_info2.set_sub_txn_id(sub_txn_id);
        sub_txn_info2.set_table_id(table_id);
        sub_txn_info2.mutable_base_tablet_ids()->Add(tablet_id2);

        req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info1));
        req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info2));
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_EQ(repair_tablet_idx_count, 2);
        ASSERT_TRUE(commit_txn_with_sub_txn_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_tablet_idx_db_id(txn, db_id, tablet_id1);
        check_tablet_idx_db_id(txn, db_id, tablet_id2);
        check_tmp_rowset_not_exist(txn, tablet_id1, txn_id);
        check_tmp_rowset_not_exist(txn, tablet_id2, sub_txn_id);
        check_rowset_meta_exist(txn, tablet_id1, 2);
        check_rowset_meta_exist(txn, tablet_id2, 2);
    }
}

// Retry tests use a real lazy publisher and real FDB transactions. Two partitions
// have two tablets each, so a retry can scan a strict subset of the original load.
class CommittedTxnRetryTest : public testing::Test {
protected:
    using Metadata = std::map<std::string, std::string>;
    static constexpr int64_t DB_ID = 93001;
    static constexpr int64_t TABLE_ID = 93002;
    static constexpr int64_t INDEX_ID = 93003;
    static constexpr int64_t PARTITION_ID = 93004;
    static constexpr int64_t TABLET_ID = 93100;
    static constexpr auto INSTANCE_ID = "test_instance";

    void SetUp() override {
        config::txn_lazy_commit_rowsets_thresold = 2;
        config::enable_cloud_parallel_txn_lazy_commit = true;
        // Both partition workers must be able to reach their synchronization point.
        config::parallel_txn_lazy_commit_num_threads = 2;
        config::cloud_txn_lazy_commit_fuzzy_possibility = 0;
        kv_ = get_fdb_txn_kv();
        service_ = get_meta_service(kv_, true);
        sp_->enable_processing();

        BeginTxnRequest begin;
        begin.set_cloud_unique_id("test_cloud_unique_id");
        auto* info = begin.mutable_txn_info();
        info->set_db_id(DB_ID);
        info->set_label("committed_txn_retry");
        info->add_table_ids(TABLE_ID);
        info->set_timeout_ms(60000);
        BeginTxnResponse response;
        brpc::Controller controller;
        service_->begin_txn(&controller, &begin, &response, nullptr);
        ASSERT_EQ(response.status().code(), MetaServiceCode::OK);
        request_.set_cloud_unique_id("test_cloud_unique_id");
        request_.set_db_id(DB_ID);
        request_.set_txn_id(response.txn_id());
        request_.set_is_2pc(false);
        request_.set_enable_txn_lazy_commit(true);

        for (int i = 0; i < 4; ++i) {
            create_tablet_with_db_id(service_.get(), DB_ID, TABLE_ID, INDEX_ID,
                                     PARTITION_ID + i / 2, TABLET_ID + i);
            auto rowset =
                    create_rowset(request_.txn_id(), TABLET_ID + i, INDEX_ID, PARTITION_ID + i / 2);
            rowset.set_num_rows(7 + i);
            rowset.set_num_segments(1);
            rowset.set_data_disk_size(100);
            rowset.set_index_disk_size(20);
            rowset.set_total_disk_size(120);
            CreateRowsetResponse result;
            prepare_rowset(service_.get(), rowset, result);
            ASSERT_EQ(result.status().code(), MetaServiceCode::OK);
            commit_rowset(service_.get(), rowset, result);
            ASSERT_EQ(result.status().code(), MetaServiceCode::OK);
            rowsets_.push_back(std::move(rowset));
        }
    }

    void TearDown() override {
        // Release workers before joining, including when a test exits via ASSERT.
        // Keep callbacks and their captured fixture state alive until all work ends.
        {
            std::lock_guard lock(mutex_);
            stop_ = true;
            cv_.notify_all();
        }
        join_requests();
        service_.reset();
        sp_->clear_all_call_backs();
        sp_->clear_trace();
        sp_->disable_processing();
        config::txn_lazy_commit_rowsets_thresold = old_threshold_;
        config::enable_cloud_parallel_txn_lazy_commit = old_parallel_;
        config::parallel_txn_lazy_commit_num_threads = old_parallel_threads_;
        config::cloud_txn_lazy_commit_fuzzy_possibility = old_fuzzy_;
    }

    CommitTxnResponse commit(const CommitTxnRequest& request) {
        CommitTxnResponse response;
        brpc::Controller controller;
        service_->commit_txn(&controller, &request, &response, nullptr);
        return response;
    }

    template <typename Predicate>
    bool await(Predicate predicate) {
        std::unique_lock lock(mutex_);
        return cv_.wait_for(lock, std::chrono::seconds(10), [&] { return stop_ || predicate(); }) &&
               !stop_;
    }

    void start_partial_lazy_commit() {
        sp_->set_call_back("TxnLazyCommitter::commit", [&](auto&& args) {
            auto partition_id = *try_any_cast<int64_t*>(args[0]);
            std::unique_lock lock(mutex_);
            ++partitions_entered_;
            cv_.notify_all();
            if (partition_id == PARTITION_ID + 1) {
                EXPECT_TRUE(cv_.wait_for(lock, std::chrono::seconds(10),
                                         [&] { return stop_ || release_last_; }));
            }
        });
        sp_->set_call_back("TxnLazyCommitTask::commit_partition::finish", [&](auto&& args) {
            if (*try_any_cast<int64_t*>(args[0]) == PARTITION_ID) {
                std::lock_guard lock(mutex_);
                first_finished_ = true;
                cv_.notify_all();
            }
        });
        sp_->set_call_back("TxnLazyCommitTask::make_committed_txn_visible::commit", [&](auto&&) {
            std::unique_lock lock(mutex_);
            before_visible_ = true;
            cv_.notify_all();
            EXPECT_TRUE(cv_.wait_for(lock, std::chrono::seconds(10),
                                     [&] { return stop_ || release_visible_; }));
        });
        sp_->set_call_back("commit_txn::wait_committed", [&](auto&& args) {
            EXPECT_EQ(*try_any_cast<int64_t*>(args[0]), request_.txn_id());
            std::lock_guard lock(mutex_);
            retry_waiting_ = true;
            cv_.notify_all();
        });
        original_ = std::thread([&] { original_response_ = commit(request_); });
    }

    void finish_partition_before_retry() {
        {
            std::lock_guard lock(mutex_);
            if (release_last_) return; // The restarted write transaction also enters this hook.
            release_last_ = true;
            cv_.notify_all();
        }
        ASSERT_TRUE(await([&] { return before_visible_; }));
        check_status(TxnStatusPB::TXN_STATUS_COMMITTED);
        // Both partitions are published now, but the final visibility transaction
        // is paused. Save versions, rowsets and all statistics before the retry.
        before_retry_ = business_metadata();
    }

    void start_retry() {
        retry_ = std::thread([&] {
            retry_response_ = commit(request_);
            std::lock_guard lock(mutex_);
            retry_done_ = true;
            cv_.notify_all();
        });
    }

    void release_visibility() {
        std::lock_guard lock(mutex_);
        release_visible_ = true;
        cv_.notify_all();
    }

    void join_requests() {
        if (original_.joinable()) original_.join();
        if (retry_.joinable()) retry_.join();
    }

    Metadata business_metadata() {
        Metadata result;
        std::unique_ptr<Transaction> txn;
        EXPECT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
        FullRangeGetOptions options;
        options.snapshot = true;
        auto iter = txn->full_range_get("\x00", "\xfe", std::move(options));
        for (auto item = iter->next(); item.has_value(); item = iter->next()) {
            // The original request legitimately changes these three keys when it
            // becomes VISIBLE. All other keys must remain identical across retry.
            if (item->first == txn_info_key({INSTANCE_ID, DB_ID, request_.txn_id()}) ||
                item->first == txn_running_key({INSTANCE_ID, DB_ID, request_.txn_id()}) ||
                item->first == recycle_txn_key({INSTANCE_ID, DB_ID, request_.txn_id()})) {
                continue;
            }
            result.emplace(item->first, item->second);
        }
        EXPECT_TRUE(iter->is_valid()) << iter->error_code();
        return result;
    }

    void check_status(TxnStatusPB expected) {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string value;
        ASSERT_EQ(txn->get(txn_info_key({INSTANCE_ID, DB_ID, request_.txn_id()}), &value),
                  TxnErrorCode::TXN_OK);
        TxnInfoPB info;
        ASSERT_TRUE(info.ParseFromString(value));
        EXPECT_EQ(info.status(), expected);
    }

    void check_visible_retry() {
        EXPECT_EQ(original_response_.status().code(), MetaServiceCode::OK);
        ASSERT_EQ(retry_response_.status().code(), MetaServiceCode::OK)
                << retry_response_.ShortDebugString();
        EXPECT_EQ(retry_response_.txn_info().status(), TxnStatusPB::TXN_STATUS_VISIBLE);
        ASSERT_EQ(retry_response_.txn_info().table_ids_size(), 1);
        EXPECT_EQ(retry_response_.txn_info().table_ids(0), TABLE_ID);
        // This is the existing FE updateVersion() cache-invalidation response:
        // VISIBLE, no partition/table versions, and no incomplete-lazy flag.
        EXPECT_TRUE(retry_response_.partition_ids().empty());
        EXPECT_TRUE(retry_response_.table_ids().empty());
        EXPECT_TRUE(retry_response_.versions().empty());
        EXPECT_TRUE(retry_response_.table_stats().empty());
        EXPECT_FALSE(retry_response_.is_lazy_commit_incomplete());
        EXPECT_EQ(business_metadata(), before_retry_);
        check_status(TxnStatusPB::TXN_STATUS_VISIBLE);
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 4; ++i) {
            std::string value;
            ASSERT_EQ(txn->get(meta_rowset_key({INSTANCE_ID, TABLET_ID + i, 2}), &value),
                      TxnErrorCode::TXN_OK);
            RowsetMetaCloudPB rowset;
            ASSERT_TRUE(rowset.ParseFromString(value));
            EXPECT_EQ(rowset.rowset_id_v2(), rowsets_[i].rowset_id_v2());
            EXPECT_EQ(rowset.start_version(), 2);
            EXPECT_EQ(rowset.end_version(), 2);
            EXPECT_EQ(rowset.num_rows(), rowsets_[i].num_rows());
            check_rowset_meta_not_exist(txn, TABLET_ID + i, 3);
            check_tmp_rowset_not_exist(txn, TABLET_ID + i, request_.txn_id());
        }
    }

    std::shared_ptr<TxnKv> kv_;
    std::unique_ptr<MetaServiceProxy> service_;
    SyncPoint* sp_ = SyncPoint::get_instance();
    CommitTxnRequest request_;
    std::vector<RowsetMetaCloudPB> rowsets_;
    Metadata before_retry_;
    CommitTxnResponse original_response_;
    CommitTxnResponse retry_response_;
    std::thread original_;
    std::thread retry_;
    std::mutex mutex_;
    std::condition_variable cv_;
    int partitions_entered_ = 0;
    bool first_finished_ = false;
    bool release_last_ = false;
    bool before_visible_ = false;
    bool release_visible_ = false;
    bool retry_waiting_ = false;
    bool retry_done_ = false;
    bool stop_ = false;

private:
    const int32_t old_threshold_ = config::txn_lazy_commit_rowsets_thresold;
    const bool old_parallel_ = config::enable_cloud_parallel_txn_lazy_commit;
    const int32_t old_parallel_threads_ = config::parallel_txn_lazy_commit_num_threads;
    const int32_t old_fuzzy_ = config::cloud_txn_lazy_commit_fuzzy_possibility;
};

TEST_F(CommittedTxnRetryTest, ImmediateRetryWaitsForVisibility) {
    // 1. Lazy publishes both partitions; pause the second partition's tmp cleanup.
    // 2. Retry scans its two tmp rowsets and selects immediate (threshold = 2).
    // 3. Finish cleanup before retry creates its write transaction, but hold the
    //    global status at COMMITTED. Retry must wait rather than publish version 3.
    start_partial_lazy_commit();
    ASSERT_TRUE(await([&] { return first_finished_ && partitions_entered_ == 2; }));
    sp_->set_call_back("commit_txn_immediately:begin",
                       [&](auto&&) { finish_partition_before_retry(); });
    start_retry();
    ASSERT_TRUE(await([&] { return retry_waiting_ || retry_done_; }));
    {
        std::lock_guard lock(mutex_);
        EXPECT_TRUE(retry_waiting_);
        EXPECT_FALSE(retry_done_);
    }
    release_visibility();
    join_requests();
    EXPECT_FALSE(retry_response_.is_lazy_commit());
    check_visible_retry();
}

TEST_F(CommittedTxnRetryTest, LazyRetryWaitsForVisibility) {
    config::txn_lazy_commit_rowsets_thresold = 1;
    start_partial_lazy_commit();
    ASSERT_TRUE(await([&] { return first_finished_ && partitions_entered_ == 2; }));
    // The same remaining two rowsets now select lazy. Remove the last pending
    // marker before it reads versions, so the existing pending-wait path cannot
    // mask a missing COMMITTED-state check.
    sp_->set_call_back("commit_txn_eventually:begin",
                       [&](auto&&) { finish_partition_before_retry(); });
    sp_->set_call_back("commit_txn_eventually::before_mark_committed", [&](auto&& args) {
        // Stop the unfixed path before its DCHECK aborts the entire test binary.
        // Reaching this point is itself the failure: a committed retry is about
        // to change commit metadata again. The immediate case reproduces the
        // actual duplicate FDB writes without injecting a conflict or error.
        ADD_FAILURE() << "A committed retry must not mark the transaction committed again";
        *try_any_cast<MetaServiceCode*>(args[0]) = MetaServiceCode::TXN_INVALID_STATUS;
        *try_any_cast<bool*>(args.back()) = true;
    });
    start_retry();
    ASSERT_TRUE(await([&] { return retry_waiting_ || retry_done_; }));
    {
        std::lock_guard lock(mutex_);
        EXPECT_TRUE(retry_waiting_);
        EXPECT_FALSE(retry_done_);
    }
    release_visibility();
    join_requests();
    EXPECT_TRUE(retry_response_.is_lazy_commit());
    check_visible_retry();
}

TEST_F(CommittedTxnRetryTest, CommittedRetryReturnsSuccessWhenLazyFails) {
    sp_->set_call_back("TxnLazyCommitTask::make_committed_txn_visible::commit", [&](auto&& args) {
        // Fail publication after the transaction has durably committed. Commit
        // retries must preserve success while reporting incomplete lazy work.
        *try_any_cast<MetaServiceCode*>(args[0]) = MetaServiceCode::KV_TXN_COMMIT_ERR;
        *try_any_cast<bool*>(args.back()) = true;
    });
    auto original = commit(request_);
    ASSERT_EQ(original.status().code(), MetaServiceCode::OK);
    ASSERT_TRUE(original.is_lazy_commit_incomplete());
    check_status(TxnStatusPB::TXN_STATUS_COMMITTED);
    auto before = business_metadata();
    auto check_incomplete_retry = [&](const CommitTxnResponse& retry) {
        ASSERT_EQ(retry.status().code(), MetaServiceCode::OK) << retry.ShortDebugString();
        EXPECT_EQ(retry.txn_info().txn_id(), request_.txn_id());
        EXPECT_EQ(retry.txn_info().status(), TxnStatusPB::TXN_STATUS_VISIBLE);
        ASSERT_EQ(retry.txn_info().table_ids_size(), 1);
        EXPECT_EQ(retry.txn_info().table_ids(0), TABLE_ID);
        // FE uses both flags to recognize incomplete publication, including
        // when this request originally selected the immediate handler.
        EXPECT_TRUE(retry.is_lazy_commit());
        EXPECT_TRUE(retry.is_lazy_commit_incomplete());
        EXPECT_TRUE(retry.partition_ids().empty());
        EXPECT_TRUE(retry.table_ids().empty());
        EXPECT_TRUE(retry.versions().empty());
        EXPECT_TRUE(retry.table_stats().empty());
        EXPECT_EQ(business_metadata(), before);
        check_status(TxnStatusPB::TXN_STATUS_COMMITTED);
    };

    // All tmp rowsets have gone, so this retry selects immediate.
    request_.set_enable_txn_lazy_commit(false);
    auto immediate_retry = commit(request_);
    check_incomplete_retry(immediate_retry);

    // Force the other entry point despite the empty tmp scan. It must return
    // the same committed-but-incomplete response when its lazy task fails.
    config::cloud_txn_lazy_commit_fuzzy_possibility = 100;
    request_.set_enable_txn_lazy_commit(true);
    auto lazy_retry = commit(request_);
    check_incomplete_retry(lazy_retry);
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithFailedLazyCommitTaskTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 67935421;
    int64_t table_id = 97432015;
    int64_t index_id = 468213;
    int64_t partition_id = 753129;
    int64_t tablet_id = 86421357;
    std::string label = "test_failed_lazy_commit_task";

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label(label);
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                             tablet_id);
    auto tmp_rowset = create_rowset(txn_id, tablet_id, index_id, partition_id);
    CreateRowsetResponse rowset_res;
    prepare_rowset(meta_service.get(), tmp_rowset, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);
    commit_rowset(meta_service.get(), tmp_rowset, rowset_res);
    ASSERT_EQ(rowset_res.status().code(), MetaServiceCode::OK);

    int32_t original_fuzzy_possibility = config::cloud_txn_lazy_commit_fuzzy_possibility;
    config::cloud_txn_lazy_commit_fuzzy_possibility = 100;
    std::atomic_bool failure_injected = false;
    auto sp = SyncPoint::get_instance();
    sp->set_call_back("convert_tmp_rowsets::before_commit", [&](auto&& args) {
        auto* code = try_any_cast<MetaServiceCode*>(args[0]);
        *code = MetaServiceCode::UNDEFINED_ERR;
        auto* pred = try_any_cast<bool*>(args.back());
        *pred = true;
        failure_injected = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        config::cloud_txn_lazy_commit_fuzzy_possibility = original_fuzzy_possibility;
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    CommitTxnRequest commit_req;
    commit_req.set_cloud_unique_id("test_cloud_unique_id");
    commit_req.set_db_id(db_id);
    commit_req.set_txn_id(txn_id);
    commit_req.set_is_2pc(false);
    commit_req.set_commit_tso(12345);
    commit_req.set_enable_txn_lazy_commit(true);
    CommitTxnResponse commit_res;
    meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                             &commit_req, &commit_res, nullptr);
    ASSERT_EQ(commit_res.status().code(), MetaServiceCode::OK);
    ASSERT_TRUE(failure_injected.load());
    ASSERT_TRUE(commit_res.has_is_lazy_commit());
    ASSERT_TRUE(commit_res.is_lazy_commit());
    ASSERT_TRUE(commit_res.has_is_lazy_commit_incomplete());
    ASSERT_TRUE(commit_res.is_lazy_commit_incomplete());

    std::unique_ptr<Transaction> txn;
    ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
    check_txn_committed(txn, db_id, txn_id, label);
    check_tmp_rowset_exist(txn, tablet_id, txn_id);
    check_rowset_meta_not_exist(txn, tablet_id, 2);
}

TEST(TxnLazyCommitTest, NotFallThroughCommitTxnEventuallyTest) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 415413556;
    int64_t table_id = 34184234;
    int64_t index_id = 9059444;
    int64_t partition_id = 8934984;
    bool commit_txn_immediatelly_hit = false;
    bool commit_txn_eventually_finish_hit = false;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_immediately::before_commit", [&](auto&& args) {
        TxnErrorCode* err = try_any_cast<TxnErrorCode*>(args[0]);
        *err = TxnErrorCode::TXN_BYTES_TOO_LARGE;

        MetaServiceCode* code = try_any_cast<MetaServiceCode*>(args[1]);
        *code = cast_as<ErrCategory::COMMIT>(*err);

        bool* pred = try_any_cast<bool*>(args.back());
        *pred = true;
        commit_txn_immediatelly_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_not_fallthrough_commit_txn_eventually");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 783426908;
    for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::INVALID_ARGUMENT);
        ASSERT_TRUE(commit_txn_immediatelly_hit);
        ASSERT_FALSE(commit_txn_eventually_finish_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
            check_rowset_meta_not_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, FallThroughCommitTxnEventuallyTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 903437831;
    int64_t table_id = 13845693;
    int64_t index_id = 2366023843;
    int64_t partition_id = 210486436;
    bool commit_txn_immediatelly_hit = false;
    bool commit_txn_eventually_finish_hit = false;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_immediately::before_commit", [&](auto&& args) {
        TxnErrorCode* err = try_any_cast<TxnErrorCode*>(args[0]);
        *err = TxnErrorCode::TXN_BYTES_TOO_LARGE;

        MetaServiceCode* code = try_any_cast<MetaServiceCode*>(args[1]);
        *code = cast_as<ErrCategory::COMMIT>(*err);

        bool* pred = try_any_cast<bool*>(args.back());
        *pred = true;
        commit_txn_immediatelly_hit = true;
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_fallthrough_commit_txn_eventually");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 1908462;
    for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString();
        ASSERT_TRUE(commit_txn_immediatelly_hit);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);
        // The failed immediate attempt already populated partition versions.
        // Falling back must return only the successful lazy attempt's results.
        ASSERT_EQ(res.partition_ids_size(), 1);
        ASSERT_EQ(res.table_ids_size(), 1);
        ASSERT_EQ(res.versions_size(), 1);
        EXPECT_EQ(res.partition_ids(0), partition_id);
        EXPECT_EQ(res.table_ids(0), table_id);
        EXPECT_EQ(res.versions(0), 2);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < config::txn_lazy_commit_rowsets_thresold; ++i) {
            int64_t tablet_id = tablet_id_base + i;

            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, ConcurrentCommitTxnEventuallyCase1Test) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //          thread1                       thread2
    //             |                             |
    //  commit_txn_eventually begin   commit_txn_eventually begin
    //             |                             |
    //       lazy commit wait                    |
    //             |                             |
    //             |                      advance last txn
    //             |                             |
    //             |                           finish
    //             |                             |
    //          finish                           |
    //             |                             |
    //             |                             |
    //             v                             v

    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 315477;
    int64_t table_id = 31752134;
    int64_t index_id = 3458532;
    int64_t partition_id = 26328765;

    std::mutex go_mutex;
    std::condition_variable go_cv;

    bool go = false;
    std::atomic<int32_t> commit_txn_eventually_begin_count = {0};
    std::atomic<int32_t> last_pending_txn_id_count = {0};
    std::atomic<int32_t> txn_lazy_committer_wait_count = {0};
    std::atomic<int32_t> finish_count = {0};

    auto sp = SyncPoint::get_instance();

    int64_t first_txn_id = 0;
    sp->set_call_back("commit_txn_eventually:begin", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        commit_txn_eventually_begin_count++;
        if (commit_txn_eventually_begin_count == 1) {
            first_txn_id = *try_any_cast<int64_t*>(args[0]);
        }
        if (commit_txn_eventually_begin_count == 2) {
            {
                go_cv.wait(_lock, [&] { return txn_lazy_committer_wait_count == 1; });
            }
        }
    });

    sp->set_call_back("commit_txn_eventually::advance_last_pending_txn_id", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        last_pending_txn_id_count++;
        if (last_pending_txn_id_count == 1) {
            int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
            ASSERT_EQ(last_pending_txn_id, first_txn_id);
        }
    });

    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_wait", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        txn_lazy_committer_wait_count++;
        if (txn_lazy_committer_wait_count == 1) {
            go_cv.notify_all();
            go_cv.wait(_lock, [&] { return finish_count == 1; });
        }
    });

    int64_t second_txn_id = 0;
    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        int64_t txn_id = *try_any_cast<int64_t*>(args[1]);
        std::unique_lock<std::mutex> _lock(go_mutex);
        finish_count++;
        if (finish_count == 1) {
            second_txn_id = txn_id;
            ASSERT_NE(second_txn_id, first_txn_id);
            go_cv.notify_all();
        }
        ASSERT_EQ(code, MetaServiceCode::OK);
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 1908462;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }

    int64_t txn_id1 = 0;
    int64_t txn1_table_version = 0;
    std::thread thread1([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually2313");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id1 = res.txn_id();
            ASSERT_GT(txn_id1, 0);
        }
        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id1, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id1);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);

            ASSERT_EQ(res.table_stats().size(), 1);
            txn1_table_version = res.table_stats()[0].table_version();
        }
    });

    int64_t txn_id2 = 0;
    int64_t txn2_table_version = 0;
    std::thread thread2([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually234142");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id2 = res.txn_id();
            ASSERT_GT(txn_id2, 0);
        }
        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id2, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id2);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);

            ASSERT_EQ(res.table_stats().size(), 1);
            txn2_table_version = res.table_stats()[0].table_version();
        }
    });

    std::unique_lock<std::mutex> go_lock(go_mutex);
    go = true;
    go_lock.unlock();
    go_cv.notify_all();

    thread1.join();
    thread2.join();

    ASSERT_TRUE(txn1_table_version == 1 && txn2_table_version == 2 ||
                txn1_table_version == 2 && txn2_table_version == 1)
            << ", txn1_table_version=" << txn1_table_version
            << ", txn2_table_version=" << txn2_table_version;

    ASSERT_EQ(commit_txn_eventually_begin_count, 3);
    ASSERT_EQ(last_pending_txn_id_count, 1);
    ASSERT_EQ(finish_count, 2);

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            std::string key = meta_tablet_idx_key({mock_instance, tablet_id});
            std::string val;
            ASSERT_EQ(txn->get(key, &val), TxnErrorCode::TXN_OK);
            TabletIndexPB tablet_idx_pb;
            tablet_idx_pb.ParseFromString(val);
            ASSERT_EQ(tablet_idx_pb.db_id(), db_id);

            std::string tmp_rowset_key =
                    meta_rowset_tmp_key({mock_instance, first_txn_id, tablet_id});
            std::string tmp_rowset_val;
            ASSERT_EQ(txn->get(tmp_rowset_key, &tmp_rowset_val), TxnErrorCode::TXN_KEY_NOT_FOUND);

            std::string rowset_key = meta_rowset_key({mock_instance, tablet_id, 2});
            std::string rowset_val;
            ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);

            tmp_rowset_key = meta_rowset_tmp_key({mock_instance, second_txn_id, tablet_id});
            tmp_rowset_val.clear();
            ASSERT_EQ(txn->get(tmp_rowset_key, &tmp_rowset_val), TxnErrorCode::TXN_KEY_NOT_FOUND);

            rowset_key = meta_rowset_key({mock_instance, tablet_id, 3});
            rowset_val.clear();
            ASSERT_EQ(txn->get(rowset_key, &rowset_val), TxnErrorCode::TXN_OK);
        }
    }
}

TEST(TxnLazyCommitTest, ConcurrentCommitTxnEventuallyCase2Test) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //           thread1                           thread2
    //              |                                 |
    //   commit_txn_eventually begin      commit_txn_immediately begin
    //              |                                 |
    //       lazy commit wait                         |
    //              |                                 |
    //              |                         advance last txn
    //              |                                 |
    //              |                               finish
    //              |                                 |
    //           finish                               |
    //              |                                 |
    //              |                                 |
    //              v                                 v

    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 134179141;
    int64_t table_id = 3243234;
    int64_t index_id = 8098324;
    int64_t partition_id = 32895321;

    std::mutex go_mutex;
    std::condition_variable go_cv;
    bool go = false;

    std::atomic<int32_t> commit_txn_immediately_begin_count = {0};
    std::atomic<int32_t> last_pending_txn_id_count = {0};
    std::atomic<int32_t> txn_lazy_committer_wait_count = {0};
    std::atomic<int32_t> immediately_finish_count = {0};
    std::atomic<int32_t> eventually_finish_count = {0};

    auto sp = SyncPoint::get_instance();

    int64_t first_txn_id = 0;
    sp->set_call_back("commit_txn_immediately:begin", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        commit_txn_immediately_begin_count++;
        if (commit_txn_immediately_begin_count == 1) {
            {
                first_txn_id = *try_any_cast<int64_t*>(args[0]);
                go_cv.wait(_lock, [&] { return txn_lazy_committer_wait_count == 1; });
            }
        }
    });

    int64_t second_txn_id = 0;
    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_wait", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        txn_lazy_committer_wait_count++;
        if (txn_lazy_committer_wait_count == 1) {
            int64_t txn_id = *try_any_cast<int64_t*>(args[0]);
            second_txn_id = txn_id;
            go_cv.notify_all();
            go_cv.wait(_lock, [&] { return immediately_finish_count == 1; });
        }
    });

    sp->set_call_back("commit_txn_immediately::advance_last_pending_txn_id", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        last_pending_txn_id_count++;
        if (last_pending_txn_id_count == 1) {
            int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
            ASSERT_EQ(last_pending_txn_id, second_txn_id);
        }
    });

    sp->set_call_back("commit_txn_immediately::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        std::unique_lock<std::mutex> _lock(go_mutex);
        immediately_finish_count++;
        if (immediately_finish_count == 1) {
            go_cv.notify_all();
        }
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        eventually_finish_count++;
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 1908462;
    for (int i = 0; i < 1999; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }

    int64_t txn_id1 = 0;
    std::thread thread1([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually3441");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id1 = res.txn_id();
            ASSERT_GT(txn_id1, 0);
        }
        {
            for (int i = 0; i < 1999; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id1, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id1);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    int64_t txn_id2 = 0;
    std::thread thread2([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually3245232");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id2 = res.txn_id();
            ASSERT_GT(txn_id2, 0);
        }
        {
            for (int i = 0; i < 1999; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id2, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id2);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(false);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    std::unique_lock<std::mutex> go_lock(go_mutex);
    go = true;
    go_lock.unlock();
    go_cv.notify_all();

    thread1.join();
    thread2.join();

    ASSERT_EQ(commit_txn_immediately_begin_count, 2);
    ASSERT_EQ(last_pending_txn_id_count, 1);
    ASSERT_EQ(immediately_finish_count, 1);
    ASSERT_EQ(eventually_finish_count, 1);

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 1999; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);

            check_tmp_rowset_not_exist(txn, tablet_id, first_txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);

            check_tmp_rowset_not_exist(txn, tablet_id, second_txn_id);
            check_rowset_meta_exist(txn, tablet_id, 3);
        }
    }
}

TEST(TxnLazyCommitTest, ConcurrentCommitTxnEventuallyCase3Test) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //            thread1              thread2
    //               |                     |
    //  commit_txn_eventually begin  get rowset begin
    //               |                     |
    //   lazy commit wait                  |
    //               |                     |
    //               |              advance last txn
    //               |                     |
    //               |                   finish
    //               |                     |
    //            finish                   |
    //               |                     |
    //               |                     |
    //               v                     v

    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 6544345;
    int64_t table_id = 32431068334;
    int64_t index_id = 132433;
    int64_t partition_id = 956120248;

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 19201262;

    for (int i = 0; i < 10001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }
    {
        int tmp_txn_id = 0;
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_32ae213dasf2");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            tmp_txn_id = res.txn_id();
            ASSERT_GT(res.txn_id(), 0);
        }

        {
            for (int i = 0; i < 10001; ++i) {
                auto tmp_rowset =
                        create_rowset(tmp_txn_id, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(tmp_txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    }

    std::mutex go_mutex;
    std::condition_variable go_cv;
    bool go = false;

    std::atomic<int32_t> get_rowset_begin_count = {0};
    std::atomic<int32_t> last_pending_txn_id_count = {0};
    std::atomic<int32_t> txn_lazy_committer_submit_count = {0};
    std::atomic<int32_t> get_rowset_finish_count = {0};
    std::atomic<int32_t> eventually_finish_count = {0};

    auto sp = SyncPoint::get_instance();

    int64_t get_rowset_tablet_id = 0;
    sp->set_call_back("get_rowset:begin", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        get_rowset_begin_count++;
        if (get_rowset_begin_count == 1) {
            {
                get_rowset_tablet_id = *try_any_cast<int64_t*>(args[0]);
                go_cv.wait(_lock, [&] { return txn_lazy_committer_submit_count == 1; });
            }
        }
    });

    int64_t eventually_txn_id = 0;
    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_submit", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        txn_lazy_committer_submit_count++;
        if (txn_lazy_committer_submit_count == 1) {
            eventually_txn_id = *try_any_cast<int64_t*>(args[0]);
            go_cv.notify_all();
            go_cv.wait(_lock, [&] { return get_rowset_finish_count == 1; });
        }
    });

    sp->set_call_back("get_rowset::advance_last_pending_txn_id", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        last_pending_txn_id_count++;
        if (last_pending_txn_id_count == 1) {
            auto txn_id = *try_any_cast<int64_t*>(args[0]);
            ASSERT_EQ(txn_id, eventually_txn_id);
        }
    });

    sp->set_call_back("get_rowset::finish", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        get_rowset_finish_count++;
        if (get_rowset_finish_count == 1) {
            go_cv.notify_all();
        }
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        eventually_finish_count++;
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    int64_t txn_id1 = 0;
    std::thread thread1([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_315322242");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id1 = res.txn_id();
            ASSERT_GT(txn_id1, 0);
        }

        {
            for (int i = 0; i < 10001; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id1, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id1);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    std::thread thread2([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            GetRowsetRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            auto tablet_idx = req.mutable_idx();
            tablet_idx->set_table_id(table_id);
            tablet_idx->set_index_id(index_id);
            tablet_idx->set_partition_id(partition_id);
            tablet_idx->set_tablet_id(tablet_id_base);
            req.set_start_version(0);
            req.set_end_version(-1);
            req.set_cumulative_compaction_cnt(0);
            req.set_base_compaction_cnt(0);
            req.set_cumulative_point(2);

            GetRowsetResponse res;
            meta_service->get_rowset(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    std::unique_lock<std::mutex> go_lock(go_mutex);
    go = true;
    go_lock.unlock();
    go_cv.notify_all();

    thread1.join();
    thread2.join();

    ASSERT_EQ(get_rowset_begin_count, 2);
    ASSERT_EQ(last_pending_txn_id_count, 1);
    ASSERT_EQ(txn_lazy_committer_submit_count, 1);
    ASSERT_EQ(get_rowset_finish_count, 1);
    ASSERT_EQ(eventually_finish_count, 1);
}

TEST(TxnLazyCommitTest, ConcurrentCommitTxnEventuallyCase4Test) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //   meta-service           recycler
    //         |                    |
    //       begin                  |
    //         |                    |
    //   lazy commit submit         |
    //         |                    |
    //       dead                   |
    //         |                    |
    //         |         abort_timeout_txn begin
    //         |                    |
    //         |             advance last txn
    //         |                    |
    //                              |
    //         |                  finish
    //         |                    |
    //         v                    v

    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 77623430234;
    int64_t table_id = 96545043;
    int64_t index_id = 2381203456;
    int64_t partition_id = 450976544;
    std::string mock_instance = "test_instance";
    const std::string label = "test_label_6787230013";

    bool commit_txn_eventullay_hit = false;
    bool abort_timeout_txn_hit = false;

    auto sp = SyncPoint::get_instance();

    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_submit", [&](auto&& args) {
        commit_txn_eventullay_hit = true;
        bool* pred = try_any_cast<bool*>(args.back());
        *pred = true;
    });

    TxnInfoPB txn_info_pb;
    sp->set_call_back("abort_timeout_txn::advance_last_pending_txn_id", [&](auto&& args) {
        abort_timeout_txn_hit = true;
        txn_info_pb = *try_any_cast<TxnInfoPB*>(args[0]);
        {
            std::unique_ptr<Transaction> txn;
            ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
            check_txn_committed(txn, db_id, txn_info_pb.txn_id(), label);
        }
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 213430076554;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }
    int txn_id = 0;
    {
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label(label);
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id = res.txn_id();
            ASSERT_GT(txn_id, 0);
        }

        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            ASSERT_TRUE(res.has_is_lazy_commit());
            ASSERT_TRUE(res.is_lazy_commit());
            ASSERT_TRUE(res.has_is_lazy_commit_incomplete());
            ASSERT_TRUE(res.is_lazy_commit_incomplete());
        }
    }

    InstanceInfoPB instance;
    instance.set_instance_id(mock_instance);
    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));

    ASSERT_EQ(recycler.init(), 0);
    ASSERT_EQ(recycler.abort_timeout_txn(), 0);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_txn_visible(txn, db_id, txn_id, label);
    }
    sleep(1);
    ASSERT_EQ(recycler.recycle_expired_txn_label(), 0);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_txn_not_exist(txn, db_id, txn_id, label);
    }

    ASSERT_TRUE(commit_txn_eventullay_hit);
    ASSERT_TRUE(abort_timeout_txn_hit);
    ASSERT_EQ(txn_id, txn_info_pb.txn_id());
}

TEST(TxnLazyCommitTest, ConcurrentCommitTxnEventuallyCase5Test) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //           thread1                           thread2
    //              |                                 |
    //   commit_txn_eventually begin      commit_txn_with_sub_txn begin
    //              |                                 |
    //       lazy commit wait                         |
    //              |                                 |
    //              |                         advance last txn
    //              |                                 |
    //              |                               finish
    //              |                                 |
    //           finish                               |
    //              |                                 |
    //              |                                 |
    //              v                                 v

    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 134179142;
    int64_t table_id = 3243264;
    int64_t index_id = 8098394;
    int64_t partition_id = 32895361;

    std::mutex go_mutex;
    std::condition_variable go_cv;
    bool go = false;

    std::atomic<int32_t> commit_txn_immediately_begin_count = {0};
    std::atomic<int32_t> last_pending_txn_id_count = {0};
    std::atomic<int32_t> txn_lazy_committer_wait_count = {0};
    std::atomic<int32_t> immediately_finish_count = {0};
    std::atomic<int32_t> eventually_finish_count = {0};

    auto sp = SyncPoint::get_instance();

    int64_t first_txn_id = 0;
    sp->set_call_back("commit_txn_with_sub_txn:begin", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        commit_txn_immediately_begin_count++;
        if (commit_txn_immediately_begin_count == 1) {
            {
                first_txn_id = *try_any_cast<int64_t*>(args[0]);
                go_cv.wait(_lock, [&] { return txn_lazy_committer_wait_count == 1; });
                go_cv.notify_all();
            }
        }
    });

    int64_t second_txn_id = 0;
    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_wait", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        txn_lazy_committer_wait_count++;
        if (txn_lazy_committer_wait_count == 1) {
            int64_t txn_id = *try_any_cast<int64_t*>(args[0]);
            second_txn_id = txn_id;
            go_cv.notify_all();
        }
    });

    sp->set_call_back("commit_txn_with_sub_txn::advance_last_pending_txn_id", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        last_pending_txn_id_count++;
        if (last_pending_txn_id_count == 1) {
            int64_t last_pending_txn_id = *try_any_cast<int64_t*>(args[0]);
            ASSERT_EQ(last_pending_txn_id, second_txn_id);
        }
        go_cv.notify_all();
    });

    sp->set_call_back("commit_txn_with_sub_txn::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        std::unique_lock<std::mutex> _lock(go_mutex);
        immediately_finish_count++;
        if (immediately_finish_count == 1) {
            go_cv.notify_all();
        }
    });

    sp->set_call_back("commit_txn_eventually::finish", [&](auto&& args) {
        MetaServiceCode code = *try_any_cast<MetaServiceCode*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        eventually_finish_count++;
    });

    sp->set_call_back("TxnLazyCommitter::commit", [&](auto&& args) {
        std::unique_lock<std::mutex> _lock(go_mutex);
        go_cv.wait(_lock, [&] { return last_pending_txn_id_count == 1; });
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 1908562;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }

    int64_t txn_id1 = 0;
    std::thread thread1([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually3442");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id1 = res.txn_id();
            ASSERT_GT(txn_id1, 0);
        }
        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id1, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id1);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    int64_t txn_id2 = 0;
    std::thread thread2([&] {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            go_cv.wait(_lock, [&] { return go; });
        }
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_concurrent_commit_txn_eventually5");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id2 = res.txn_id();
            ASSERT_GT(txn_id2, 0);
        }
        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset =
                        create_rowset(txn_id2, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }
        int64_t sub_txn_id1 = txn_id2;

        // begin sub_txn1
        int64_t sub_txn_id2 = -1;
        {
            brpc::Controller cntl;
            BeginSubTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_txn_id(txn_id2);
            req.set_sub_txn_num(0);
            req.set_db_id(db_id);
            req.set_label("test_label_concurrent_commit_txn_eventually5_sub");
            req.mutable_table_ids()->Add(table_id);
            req.mutable_table_ids()->Add(table_id);
            BeginSubTxnResponse res;
            meta_service->begin_sub_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                        &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            ASSERT_EQ(res.txn_info().table_ids().size(), 2);
            ASSERT_EQ(res.txn_info().sub_txn_ids().size(), 1);
            ASSERT_TRUE(res.has_sub_txn_id());
            sub_txn_id2 = res.sub_txn_id();
            ASSERT_EQ(sub_txn_id2, res.txn_info().sub_txn_ids()[0]);
        }
        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset =
                        create_rowset(sub_txn_id2, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id2);
            req.set_is_txn_load(true);

            SubTxnInfo sub_txn_info1;
            sub_txn_info1.set_sub_txn_id(sub_txn_id1);
            sub_txn_info1.set_table_id(table_id);
            for (int i = 0; i < 2001; ++i) {
                sub_txn_info1.mutable_base_tablet_ids()->Add(tablet_id_base + i);
            }

            SubTxnInfo sub_txn_info2;
            sub_txn_info2.set_sub_txn_id(sub_txn_id2);
            sub_txn_info2.set_table_id(table_id);
            for (int i = 0; i < 2001; ++i) {
                sub_txn_info1.mutable_base_tablet_ids()->Add(tablet_id_base + i);
            }

            req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info1));
            req.mutable_sub_txn_infos()->Add(std::move(sub_txn_info2));
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    });

    std::unique_lock<std::mutex> go_lock(go_mutex);
    go = true;
    go_lock.unlock();
    go_cv.notify_all();

    thread1.join();
    thread2.join();

    ASSERT_EQ(commit_txn_immediately_begin_count, 2);
    ASSERT_EQ(last_pending_txn_id_count, 1);
    ASSERT_EQ(immediately_finish_count, 1);
    ASSERT_EQ(eventually_finish_count, 1);

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);

            check_tmp_rowset_not_exist(txn, tablet_id, first_txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);

            check_tmp_rowset_not_exist(txn, tablet_id, second_txn_id);
            check_rowset_meta_exist(txn, tablet_id, 4);
        }
    }
}

TEST(TxnLazyCommitTest, RowsetMetaSizeExceedTest) {
    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 5252025;
    int64_t table_id = 35201043384;
    int64_t index_id = 256439;
    int64_t partition_id = 732536259;

    auto meta_service = get_meta_service(txn_kv, true);
    int64_t tablet_id = 25910248;

    {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id);
    }
    {
        int tmp_txn_id = 0;
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label("test_label_32ae213dasg3");
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            tmp_txn_id = res.txn_id();
            ASSERT_GT(res.txn_id(), 0);
        }
        {
            auto tmp_rowset = create_rowset(tmp_txn_id, tablet_id, index_id, partition_id);
            CreateRowsetResponse res;
            prepare_rowset(meta_service.get(), tmp_rowset, res);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            commit_rowset(meta_service.get(), tmp_rowset, res);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(tmp_txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    }

    auto* sp = SyncPoint::get_instance();
    sp->set_call_back("get_rowset:meta_exceed_limit", [](auto&& args) {
        auto* byte_size = try_any_cast<size_t*>(args[0]);
        *byte_size = std::numeric_limits<int32_t>::max();
        ++(*byte_size);
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };
    {
        brpc::Controller cntl;
        GetRowsetRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        auto* tablet_idx = req.mutable_idx();
        tablet_idx->set_table_id(table_id);
        tablet_idx->set_index_id(index_id);
        tablet_idx->set_partition_id(partition_id);
        tablet_idx->set_tablet_id(tablet_id);
        req.set_start_version(0);
        req.set_end_version(-1);
        req.set_cumulative_compaction_cnt(0);
        req.set_base_compaction_cnt(0);
        req.set_cumulative_point(2);

        GetRowsetResponse res;
        meta_service->get_rowset(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::PROTOBUF_PARSE_ERR);
    }
}

TEST(TxnLazyCommitTest, ForceTxnLazyCommit) {
    int counter = 0;
    config::cloud_txn_lazy_commit_fuzzy_possibility = 0;
    for (size_t i = 0; i < 100000; i++) {
        if (force_txn_lazy_commit()) {
            counter++;
        }
    }
    LOG(INFO) << "force_txn_lazy_commit counter: " << counter;
    ASSERT_EQ(counter, 0);

    config::cloud_txn_lazy_commit_fuzzy_possibility = 50;
    counter = 0;
    for (size_t i = 0; i < 100000; i++) {
        if (force_txn_lazy_commit()) {
            counter++;
        }
    }
    LOG(INFO) << "force_txn_lazy_commit counter: " << counter;
    ASSERT_GT(counter, 30000);
    ASSERT_LT(counter, 70000);
    config::cloud_txn_lazy_commit_fuzzy_possibility = 0;
}

TEST(TxnLazyCommitTest, RecyclePartitions) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //   meta-service             recycler
    //         |                      |
    //       begin                    |
    //         |                      |
    // prepare/commit rowset          |
    //         |                      |
    //   lazy commit submit           |
    //         |                      |
    //       dead                     |
    //         |                      |
    //   drop partition               |
    //         |                      |
    //         |                recyle_partitions
    //         |                      |
    //         |                      |
    //         |                      |
    //         |                      |
    //         v                      v

    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 988032131;
    int64_t table_id = 5145043;
    int64_t index_id = 273456;
    int64_t partition_id = 44576544;
    std::string mock_instance = "test_instance";
    const std::string label = "test_label_67ae2q1231";

    bool commit_txn_eventullay_hit = false;
    bool txn_not_finished_hit = false;

    auto sp = SyncPoint::get_instance();

    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_submit", [&](auto&& args) {
        commit_txn_eventullay_hit = true;
        bool* pred = try_any_cast<bool*>(args.back());
        *pred = true;
    });

    sp->set_call_back("check_lazy_txn_finished::txn_not_finished",
                      [&](auto&& args) { txn_not_finished_hit = true; });

    sp->set_call_back("recycle_tablet::begin", [](auto&& args) {
        auto* ret = try_any_cast_ret<int>(args);
        ret->first = 0;
        ret->second = true;
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 2313324;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }
    int txn_id = 0;
    {
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label(label);
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id = res.txn_id();
            ASSERT_GT(txn_id, 0);
        }

        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }

        {
            // drop partition
            brpc::Controller cntl;
            PartitionRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");

            req.set_db_id(1);
            req.set_table_id(table_id);
            req.add_index_ids(index_id);
            req.add_partition_ids(partition_id);
            PartitionResponse res;
            meta_service->drop_partition(
                    reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                    nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    }

    InstanceInfoPB instance;
    instance.set_instance_id(mock_instance);
    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));

    ASSERT_EQ(recycler.init(), 0);
    ASSERT_FALSE(txn_not_finished_hit);
    ASSERT_EQ(recycler.recycle_partitions(), -1);
    ASSERT_TRUE(txn_not_finished_hit);

    ASSERT_EQ(recycler.abort_timeout_txn(), 0);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_txn_visible(txn, db_id, txn_id, label);
    }

    ASSERT_EQ(recycler.recycle_partitions(), 0);
}

TEST(TxnLazyCommitTest, RecycleIndexes) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //   meta-service             recycler
    //         |                      |
    //       begin                    |
    //         |                      |
    // prepare/commit rowset          |
    //         |                      |
    //   lazy commit submit           |
    //         |                      |
    //       dead                     |
    //         |                      |
    //   drop indexes                 |
    //         |                      |
    //         |                recycle_indexes
    //         |                      |
    //         |                      |
    //         |                      |
    //         |                      |
    //         v                      v

    auto txn_kv = get_mem_txn_kv();

    int64_t db_id = 4524364;
    int64_t table_id = 65354;
    int64_t index_id = 658432;
    int64_t partition_id = 76553;
    std::string mock_instance = "test_instance";
    const std::string label = "test_label_67a34e2q1231";

    bool commit_txn_eventullay_hit = false;
    bool txn_not_finished_hit = false;

    auto sp = SyncPoint::get_instance();

    sp->set_call_back("commit_txn_eventually::txn_lazy_committer_submit", [&](auto&& args) {
        commit_txn_eventullay_hit = true;
        bool* pred = try_any_cast<bool*>(args.back());
        *pred = true;
    });

    sp->set_call_back("check_lazy_txn_finished::txn_not_finished",
                      [&](auto&& args) { txn_not_finished_hit = true; });

    sp->set_call_back("recycle_tablet::begin", [](auto&& args) {
        auto* ret = try_any_cast_ret<int>(args);
        ret->first = 0;
        ret->second = true;
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    // mock rowset and tablet
    int64_t tablet_id_base = 2313324;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
    }
    int txn_id = 0;
    {
        {
            brpc::Controller cntl;
            BeginTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            TxnInfoPB txn_info_pb;
            txn_info_pb.set_db_id(db_id);
            txn_info_pb.set_label(label);
            txn_info_pb.add_table_ids(table_id);
            txn_info_pb.set_timeout_ms(36000);
            req.mutable_txn_info()->CopyFrom(txn_info_pb);
            BeginTxnResponse res;
            meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                    &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            txn_id = res.txn_id();
            ASSERT_GT(txn_id, 0);
        }

        {
            for (int i = 0; i < 2001; ++i) {
                auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
                CreateRowsetResponse res;
                prepare_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                commit_rowset(meta_service.get(), tmp_rowset, res);
                ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            }
        }

        {
            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }

        {
            // drop partition
            brpc::Controller cntl;
            IndexRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");

            req.set_db_id(1);
            req.set_table_id(table_id);
            req.add_index_ids(index_id);
            IndexResponse res;
            meta_service->drop_index(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    }

    InstanceInfoPB instance;
    instance.set_instance_id(mock_instance);
    InstanceRecycler recycler(txn_kv, instance, thread_group,
                              std::make_shared<TxnLazyCommitter>(txn_kv));

    ASSERT_EQ(recycler.init(), 0);
    ASSERT_FALSE(txn_not_finished_hit);
    ASSERT_EQ(recycler.recycle_indexes(), -1);
    ASSERT_TRUE(txn_not_finished_hit);

    ASSERT_EQ(recycler.abort_timeout_txn(), 0);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        check_txn_visible(txn, db_id, txn_id, label);
    }

    ASSERT_EQ(recycler.recycle_indexes(), 0);
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithMultiTableTest) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 3132121;
    int64_t table_id = 5452432;
    int64_t index_id = 76763;
    int64_t partition_id = 43432;

    int64_t table_id2 = 54524321231;
    int64_t index_id2 = 543123;
    int64_t partition_id2 = 214352;
    bool commit_txn_eventually_finish_hit = false;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::task->wait", [&](auto&& args) {
        auto [code, msg] = *try_any_cast<std::pair<MetaServiceCode, std::string>*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });
    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_multi_table_commit_txn");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.add_table_ids(table_id2);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 3131124;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    for (int i = 2001; i < 4002; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id2, index_id2, partition_id2,
                                 tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id2, partition_id2);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 4002; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithHugeRowsetMetaTest) {
    auto txn_kv = get_fdb_txn_kv();
    int64_t db_id = 14135425;
    int64_t table_id = 31245456;
    int64_t index_id = 434324;
    int64_t partition_id = 3215764;

    int64_t table_id2 = 213476;
    int64_t index_id2 = 126765;
    int64_t partition_id2 = 214567;
    bool commit_txn_eventually_finish_hit = false;

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("commit_txn_eventually::task->wait", [&](auto&& args) {
        auto [code, msg] = *try_any_cast<std::pair<MetaServiceCode, std::string>*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });

    sp->set_call_back("TxnLazyCommitTask::commit::max_rowsets_per_batch", [&](auto&& args) {
        size_t max_rowsets_per_batch = *try_any_cast<size_t*>(args[0]);
        size_t max_rowset_meta_size = *try_any_cast<size_t*>(args[1]);
        LOG(INFO) << "max_rowsets_per_batch:" << max_rowsets_per_batch
                  << " max_rowset_meta_size:" << max_rowset_meta_size;
        ASSERT_EQ(max_rowsets_per_batch, 134);
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_with_huge_rowsetmeta_test");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.add_table_ids(table_id2);
    txn_info_pb.set_timeout_ms(600000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString();
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 3131124;
    for (int i = 0; i < 1001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_huge_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    for (int i = 1001; i < 2002; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id2, index_id2, partition_id2,
                                 tablet_id_base + i);
        auto tmp_rowset = create_huge_rowset(txn_id, tablet_id_base + i, index_id2, partition_id2);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(commit_txn_eventually_finish_hit);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 2002; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_not_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithSchemaChangeTest) {
    // ===========================================================================
    // threads concurrent execution flow:
    //
    //            thread1             lazy thread1                          thread3
    //               |                     |                                   |
    //  commit_txn_eventually begin        |                                   |
    //               |                     |                                   |
    //   lazy commit wait                  |                                   |
    //               |                     |                                   |
    //               |              make_committed_txn_visible                 |
    //               |                     |                                   |
    //               |              inject TXN_TOO_OLD fdb error               |
    //               |                     |                      sc create new tablet tmp rowset
    //               |                     |                                   |
    //               |                     |                                   |
    //       retry commit_txn              |                                   |
    //               v                     v
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 4534445675;
    int64_t table_id = 4365676543;
    int64_t index_id = 665453237;
    int64_t partition_id = 2136776543678;

    bool go = false;
    std::mutex go_mutex;
    std::condition_variable go_cv;
    std::atomic<int32_t> make_committed_txn_visible_count = {0};
    std::atomic<int32_t> sc_create_tmp_rowset_count = {0};
    std::atomic<int32_t> sc_create_tmp_rowset_finish_count = {0};
    std::atomic<int32_t> tmp_rowsets_been_already_converted = {0};

    auto sp = SyncPoint::get_instance();
    sp->set_call_back("TxnLazyCommitTask::make_committed_txn_visible::commit", [&](auto&& args) {
        {
            std::unique_lock<std::mutex> _lock(go_mutex);
            if (make_committed_txn_visible_count == 0) {
                make_committed_txn_visible_count++;
                if (sc_create_tmp_rowset_count == 0) {
                    go_cv.wait(_lock, [&] { return sc_create_tmp_rowset_count == 1; });
                }
                MetaServiceCode* code = try_any_cast<MetaServiceCode*>(args[0]);
                *code = MetaServiceCode::KV_TXN_CONFLICT;
                bool* pred = try_any_cast<bool*>(args.back());
                *pred = true;
                LOG(INFO) << "inject kv error KV_TXN_CONFLICT";
                go_cv.notify_all();
            }
        }
    });

    sp->set_call_back("convert_tmp_rowsets::already_been_converted", [&](auto&& args) {
        auto version_pb = *try_any_cast<VersionPB*>(args[0]);
        LOG(INFO) << "version_pb:" << version_pb.ShortDebugString();
        std::unique_lock<std::mutex> _lock(go_mutex);
        tmp_rowsets_been_already_converted++;
        go_cv.notify_all();
    });

    sp->set_call_back("TxnLazyCommitTask::commit::already_been_converted", [&](auto&& args) {
        auto version_pb = *try_any_cast<VersionPB*>(args[0]);
        LOG(INFO) << "version_pb:" << version_pb.ShortDebugString();
        std::unique_lock<std::mutex> _lock(go_mutex);
        tmp_rowsets_been_already_converted++;
        go_cv.notify_all();
    });

    sp->enable_processing();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_sc_with_commit_txn_label");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(600000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet
    int64_t tablet_id_base = 3131124;
    for (int i = 0; i < 1001; ++i) {
        create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                 tablet_id_base + i);
        auto tmp_rowset = create_huge_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    std::thread thread1([&] {
        {
            {
                std::unique_lock<std::mutex> _lock(go_mutex);
                go_cv.wait(_lock, [&] { return go; });
            }

            brpc::Controller cntl;
            CommitTxnRequest req;
            req.set_cloud_unique_id("test_cloud_unique_id");
            req.set_db_id(db_id);
            req.set_txn_id(txn_id);
            req.set_is_2pc(false);
            req.set_enable_txn_lazy_commit(true);
            CommitTxnResponse res;
            meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl),
                                     &req, &res, nullptr);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
        LOG(INFO) << "thread1 finish";
    });

    std::thread thread2([&] {
        {
            {
                std::unique_lock<std::mutex> _lock(go_mutex);
                go_cv.wait(_lock, [&] { return go; });
            }

            {
                std::unique_lock<std::mutex> _lock(go_mutex);
                sc_create_tmp_rowset_count++;
                if (make_committed_txn_visible_count == 0) {
                    go_cv.wait(_lock, [&] { return make_committed_txn_visible_count > 0; });
                }
                for (int i = 0; i < 1001; ++i) {
                    create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id,
                                             partition_id, tablet_id_base + i);
                    auto tmp_rowset =
                            create_huge_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
                    CreateRowsetResponse res;
                    prepare_rowset(meta_service.get(), tmp_rowset, res);
                    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                    commit_rowset(meta_service.get(), tmp_rowset, res);
                    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
                }
                LOG(INFO) << "sc_create_tmp_rowset_finish_count finish";
                sc_create_tmp_rowset_finish_count++;
                go_cv.notify_all();
            }
            LOG(INFO) << "thread2 finish";
        }
    });

    std::unique_lock<std::mutex> go_lock(go_mutex);
    go = true;
    go_lock.unlock();
    go_cv.notify_all();

    thread1.join();
    thread2.join();

    ASSERT_GE(tmp_rowsets_been_already_converted, 1);
    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        std::string mock_instance = "test_instance";
        for (int i = 0; i < 1001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithAbortAfterCommitTest) {
    auto txn_kv = get_mem_txn_kv();
    int64_t db_id = 55432134;
    int64_t table_id = 326843;
    int64_t index_id = 34345678;
    int64_t partition_id = 212343;

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_commit_txn_eventually");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(36000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    int64_t txn_id = res.txn_id();
    auto sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };
    sp->enable_processing();
    sp->set_call_back("commit_txn_eventually::abort_txn_after_mark_txn_commited", [&](auto&&) {
        AbortTxnRequest req;
        AbortTxnResponse res;
        req.set_txn_id(txn_id);
        req.set_cloud_unique_id("test_cloud_unique_id");
        meta_service->abort_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                &res, nullptr);
        ASSERT_EQ(res.status().actual_code(), MetaServiceCode::TXN_ALREADY_COMMITED);
    });

    // mock rowset and tablet
    int64_t tablet_id_base = 372323;
    for (int i = 0; i < 2001; ++i) {
        create_tablet_without_db_id(meta_service.get(), table_id, index_id, partition_id,
                                    tablet_id_base + i);
        auto tmp_rowset = create_rowset(txn_id, tablet_id_base + i, index_id, partition_id);
        CreateRowsetResponse res;
        prepare_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        commit_rowset(meta_service.get(), tmp_rowset, res);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tmp_rowset_exist(txn, tablet_id, txn_id);
        }
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            req.add_base_tablet_ids(tablet_id);
        }
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(res.has_is_lazy_commit());
        ASSERT_TRUE(res.is_lazy_commit());
        ASSERT_TRUE(res.has_is_lazy_commit_incomplete());
        ASSERT_FALSE(res.is_lazy_commit_incomplete());
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            req.add_base_tablet_ids(tablet_id);
        }
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        ASSERT_TRUE(res.has_is_lazy_commit());
        ASSERT_FALSE(res.is_lazy_commit());
        ASSERT_FALSE(res.has_is_lazy_commit_incomplete());
        ASSERT_FALSE(res.is_lazy_commit_incomplete());
    }

    {
        std::unique_ptr<Transaction> txn;
        ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK);
        for (int i = 0; i < 2001; ++i) {
            int64_t tablet_id = tablet_id_base + i;
            check_tablet_idx_db_id(txn, db_id, tablet_id);
            check_rowset_meta_exist(txn, tablet_id, 2);
        }
    }
}

TEST(TxnLazyCommitTest, CommitTxnEventuallyWithManyPartitions) {
    auto txn_kv = get_fdb_txn_kv();
    int64_t db_id = 15246789;
    int64_t table_id = 42356789;
    int64_t index_id = 98765432;

    std::atomic_bool commit_txn_eventually_finish_hit = false;

    auto sp = SyncPoint::get_instance();
    DORIS_CLOUD_DEFER {
        sp->clear_all_call_backs();
        sp->clear_trace();
        sp->disable_processing();
    };
    sp->set_call_back("commit_txn_eventually::task->wait", [&](auto&& args) {
        auto [code, msg] = *try_any_cast<std::pair<MetaServiceCode, std::string>*>(args[0]);
        ASSERT_EQ(code, MetaServiceCode::OK);
        commit_txn_eventually_finish_hit = true;
    });

    sp->enable_processing();

    auto meta_service = get_meta_service(txn_kv, true);
    brpc::Controller cntl;
    BeginTxnRequest req;
    req.set_cloud_unique_id("test_cloud_unique_id");
    TxnInfoPB txn_info_pb;
    txn_info_pb.set_db_id(db_id);
    txn_info_pb.set_label("test_label_with_many_partitions");
    txn_info_pb.add_table_ids(table_id);
    txn_info_pb.set_timeout_ms(60000000);
    req.mutable_txn_info()->CopyFrom(txn_info_pb);
    BeginTxnResponse res;
    meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, &res,
                            nullptr);
    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString();
    int64_t txn_id = res.txn_id();

    // mock rowset and tablet: 2000 partitions, 64 tablets per partition
    int64_t tablet_id_base = 1000000;
    int64_t partition_id_base = 5000000;

    // Change below parameters to large values if you want to do a more stress test
    const int num_partitions = 20;
    const int tablets_per_partition = 2;

    for (int p = 0; p < num_partitions; ++p) {
        int64_t partition_id = partition_id_base + p;
        for (int t = 0; t < tablets_per_partition; ++t) {
            int64_t tablet_id = tablet_id_base + p * tablets_per_partition + t;
            create_tablet_with_db_id(meta_service.get(), db_id, table_id, index_id, partition_id,
                                     tablet_id);
            auto tmp_rowset = create_rowset(txn_id, tablet_id, index_id, partition_id);
            CreateRowsetResponse res;
            prepare_rowset(meta_service.get(), tmp_rowset, res);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
            commit_rowset(meta_service.get(), tmp_rowset, res);
            ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
        }
    }

    {
        brpc::Controller cntl;
        CommitTxnRequest req;
        req.set_cloud_unique_id("test_cloud_unique_id");
        req.set_db_id(db_id);
        req.set_txn_id(txn_id);
        req.set_is_2pc(false);
        req.set_enable_txn_lazy_commit(true);
        CommitTxnResponse res;
        meta_service->commit_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req,
                                 &res, nullptr);
        ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
    }

    for (size_t i = 0; i < 1000; i++) {
        // just wait for a while to make sure the commit txn eventually task is finished
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        if (commit_txn_eventually_finish_hit.load()) {
            break;
        }
    }
}

} // namespace doris::cloud