nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 17:52:47 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
doris/cloud/test/txn_lazy_commit_test.cpp
T
bobhan1 97ae75cb12 [fix](cloud) Resume committed transactions through lazy publication (#69606) 
### What problem does this PR solve?

Related PR: #67989

A transaction commit retry can scan the temporary rowsets left by
concurrent lazy publication, then start its write transaction after
those rowsets and their pending markers have been removed. The
transaction is still COMMITTED, so processing the old scan can publish
the same rowsets again or return only part of the transaction's
partition versions.

Handle COMMITTED explicitly in the immediate, eventual and
sub-transaction commit paths through a shared wait_for_committed_txn
helper. After the non-snapshot txn_info read in the actual write
transaction, release that transaction and submit or join the lazy task.
After publication succeeds, restart and return the durable VISIBLE
transaction through the existing idempotent branch, with empty version
results so FE invalidates its version caches. A concurrent transition
from PREPARED to COMMITTED conflicts with the same txn_info read at KV
commit.

If the lazy task fails, the transaction has already committed and the
RPC still returns OK. Match the existing lazy response contract: return
txn_info as VISIBLE for FE callbacks, with is_lazy_commit=true and
is_lazy_commit_incomplete=true, without changing the durable COMMITTED
state or publishing the old scan. The two flags also identify incomplete
publication when the retry originally selected immediate, so FE does not
treat it as completed visibility.

Also clear the response before falling back from a failed immediate
attempt to lazy commit. Immediate may populate version results before
its KV commit fails with TXN_BYTES_TOO_LARGE; those results must not be
combined with the successful lazy response or retained in an
already-visible response.

#### Commit structure

1. Add three deterministic real-FDB retry tests and strengthen the
existing fallback response test.
2. Resume COMMITTED transactions through lazy publication in all three
commit paths, preserving commit success if publication fails.
3. Clear failed immediate response data before lazy fallback.

#### Validation

- Real-FDB retry tests cover an immediate retry with a stale scan, a
retry that still selects lazy, and publication failure after the
transaction has committed. The failure case checks both immediate and
lazy retry responses: OK, incomplete-lazy flags, no version/statistics
hints, unchanged business metadata and durable COMMITTED status.
- The immediate regression reproduced real version-3 duplicates on the
base implementation. The lazy regression fails before its existing
DCHECK aborts the whole test binary. The existing MemTxnKv fallback test
rejects duplicate version results from a failed immediate attempt.
- Final focused ASAN run: **all four related cases passed, zero
failures/errors**. A complete-file rerun exited 137 during
ConcurrentCommitTxnEventuallyCase3Test before producing an XML report;
no assertion failure was logged before interruption, and this run is not
counted as passing. FDB 7.1.57 was deployed as a private instance with
the repository's tools/fdb/fdb_ctl.sh; its processes were stopped and
temporary configuration restored afterward.
- Changed-line clang-format 16 checks and git diff --check passed. The
build-hygiene check reports an existing BE util/threadpool.h reach
overrun (1088 TUs versus a limit of 1087), also reproduced from archived
HEAD sources before this refactor; no BE files were changed. Clang-tidy
was not run.

The successful retry cases use bounded gates to synchronize actual
partition cleanup and final visibility, compare business metadata
including statistics, check rowset identities and versions, and verify
the empty VISIBLE response used by FE cache invalidation. These are MS
tests; the failure case does not run an FE process or independently
reproduce the sub-transaction COMMITTED retry timeline.

### Release note

Prevent duplicate publication and incomplete version responses when
transaction commit retries race lazy publication, while preserving
commit success when lazy publication is incomplete.

### Check List (For Author)

- Test
    - [x] Unit Test
- Behavior changed:
- [x] Yes. Resume lazy publication for COMMITTED retries, preserve
commit success on publication failure, and discard failed immediate
response data before fallback.
- Does this need documentation?
    - [x] No. No new user-facing setting or syntax.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
2026-10-27 10:10:24 +08:00

4215 lines
165 KiB
C++
Raw Permalink Blame History

// 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
Reference in new issue View Git Blame Copy Permalink