This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 18d97ed8463 branch-4.2: [fix](cloud) Resume committed transactions
through lazy publication #69606 (#69656)
18d97ed8463 is described below
commit 18d97ed84630cd404b45d433f84d697c39a8f19a
Author: bobhan1 <[email protected]>
AuthorDate: Tue Oct 27 19:23:25 2026 +0800
branch-4.2: [fix](cloud) Resume committed transactions through lazy
publication #69606 (#69656)
### What problem does this PR solve?
Related PR: #69606
Backport #69606 to `branch-4.2` from commit
`97ae75cb12f532df0ee42da4bc288559bd19b8a3`.
When a commit retry observes a transaction that is already COMMITTED,
the immediate, eventual and sub-transaction paths now join or resume
lazy publication and reread the durable VISIBLE state. This prevents a
stale temporary-rowset scan from publishing duplicate rowsets or
returning partial partition versions. Publication failure preserves
commit success with the existing incomplete-lazy response flags. Failed
immediate response data is cleared before lazy fallback.
Context conflicts were resolved against the existing 4.2 transaction
flow. The added and removed lines in all four files match the upstream
change.
### Validation
- ASAN Cloud UT build passed: `BUILD_TYPE=ASAN ./run-cloud-ut.sh -j100`.
- `txn_lazy_commit_test` with filter
`CommittedTxnRetryTest.*:TxnLazyCommitTest.FallThroughCommitTxnEventuallyTest`:
**4 passed, 0 failures, 0 errors**, verified in the GTest XML report.
The three retry cases used a private FoundationDB 7.1.26 instance; the
fallback case uses MemTxnKv.
- Changed-line clang-format 16 and `git diff --check` passed.
### 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 COMMITTED retries through lazy publication 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
---
cloud/src/meta-service/meta_service.h | 4 +
cloud/src/meta-service/meta_service_txn.cpp | 55 ++++
cloud/src/meta-service/txn_lazy_committer.cpp | 3 +-
cloud/test/txn_lazy_commit_test.cpp | 359 ++++++++++++++++++++++++++
4 files changed, 420 insertions(+), 1 deletion(-)
diff --git a/cloud/src/meta-service/meta_service.h
b/cloud/src/meta-service/meta_service.h
index 777cc581d7c..d002d55ba9e 100644
--- a/cloud/src/meta-service/meta_service.h
+++ b/cloud/src/meta-service/meta_service.h
@@ -481,6 +481,10 @@ private:
bool is_version_read_enabled(std::string_view instance_id) const;
bool is_version_write_enabled(std::string_view instance_id) const;
+ bool wait_for_committed_txn(const std::string& instance_id, const
TxnInfoPB& txn_info,
+ std::unique_ptr<Transaction>& txn,
CommitTxnResponse* response,
+ MetaServiceCode& code, std::string& msg,
KVStats& stats);
+
void commit_txn_immediately(
const CommitTxnRequest* request, CommitTxnResponse* response,
MetaServiceCode& code,
std::string& msg, const std::string& instance_id, int64_t db_id,
diff --git a/cloud/src/meta-service/meta_service_txn.cpp
b/cloud/src/meta-service/meta_service_txn.cpp
index 3e6026ec35a..91d7d90709e 100644
--- a/cloud/src/meta-service/meta_service_txn.cpp
+++ b/cloud/src/meta-service/meta_service_txn.cpp
@@ -1515,6 +1515,36 @@ std::pair<MetaServiceCode, std::string>
get_partition_versions(
return {MetaServiceCode::OK, ""};
}
+// Returns true when the caller should reread VISIBLE.
+bool MetaServiceImpl::wait_for_committed_txn(const std::string& instance_id,
+ const TxnInfoPB& txn_info,
+ std::unique_ptr<Transaction>& txn,
+ CommitTxnResponse* response,
MetaServiceCode& code,
+ std::string& msg, KVStats& stats)
{
+ // Lazy publication owns this transaction; do not reuse its stale rowset
scan.
+ stats.get_bytes += txn->get_bytes();
+ stats.get_counter += txn->num_get_keys();
+ txn.reset();
+ auto txn_id = txn_info.txn_id();
+ auto task = txn_lazy_committer_->submit(instance_id, txn_id);
+ TEST_SYNC_POINT_CALLBACK("commit_txn::wait_committed", &txn_id);
+ std::tie(code, msg) = task->wait();
+ if (code == MetaServiceCode::OK) {
+ return true;
+ }
+
+ LOG(WARNING) << "txn lazy commit failed txn_id=" << txn_id << " code=" <<
code
+ << " msg=" << msg;
+ // The transaction is committed; match the incomplete lazy response
contract.
+ code = MetaServiceCode::OK;
+ msg.clear();
+ response->mutable_txn_info()->CopyFrom(txn_info);
+ response->mutable_txn_info()->set_status(TxnStatusPB::TXN_STATUS_VISIBLE);
+ response->set_is_lazy_commit(true);
+ response->set_is_lazy_commit_incomplete(true);
+ return false;
+}
+
/**
* 0. Extract txn_id from request
* 1. Get db id from TxnKv with txn_id
@@ -1628,6 +1658,13 @@ void MetaServiceImpl::commit_txn_immediately(
return;
}
+ if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) {
+ if (!wait_for_committed_txn(instance_id, txn_info, txn, response,
code, msg, stats)) {
+ return;
+ }
+ continue;
+ }
+
LOG(INFO) << "txn_id=" << txn_id << " txn_info=" <<
txn_info.ShortDebugString();
CloneChainReader meta_reader(instance_id, resource_mgr_.get());
@@ -2454,6 +2491,13 @@ void MetaServiceImpl::commit_txn_eventually(
return;
}
+ if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) {
+ if (!wait_for_committed_txn(instance_id, txn_info, txn, response,
code, msg, stats)) {
+ return;
+ }
+ continue;
+ }
+
auto now_time = system_clock::now();
uint64_t commit_time =
duration_cast<milliseconds>(now_time.time_since_epoch()).count();
if ((txn_info.prepare_time() + txn_info.timeout_ms()) < commit_time) {
@@ -2468,6 +2512,7 @@ void MetaServiceImpl::commit_txn_eventually(
if (request->has_commit_attachment()) {
txn_info.mutable_commit_attachment()->CopyFrom(request->commit_attachment());
}
+
TEST_SYNC_POINT_RETURN_WITH_VOID("commit_txn_eventually::before_mark_committed",
&code);
DCHECK(txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED);
// set status TXN_STATUS_COMMITTED not TXN_STATUS_VISIBLE !!!
// lazy commit task will advance txn to make txn visible
@@ -2851,6 +2896,13 @@ void MetaServiceImpl::commit_txn_with_sub_txn(const
CommitTxnRequest* request,
return;
}
+ if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) {
+ if (!wait_for_committed_txn(instance_id, txn_info, txn, response,
code, msg, stats)) {
+ return;
+ }
+ continue;
+ }
+
LOG(INFO) << "txn_id=" << txn_id << " txn_info=" <<
txn_info.ShortDebugString();
AnnotateTag txn_tag("txn_id", txn_id);
@@ -3425,6 +3477,9 @@ void
MetaServiceImpl::commit_txn(::google::protobuf::RpcController* controller,
<< " tmp_rowsets_meta.size=" << tmp_rowsets_meta.size();
code = MetaServiceCode::OK;
msg.clear();
+ // Drop failed immediate results so FE's
CloudGlobalTransactionMgr.updateVersion()
+ // can invalidate caches when lazy returns only VISIBLE txn_info.
+ response->Clear();
commit_txn_eventually(request, response, code, msg, instance_id, db_id,
tmp_rowsets_meta,
stats);
}
diff --git a/cloud/src/meta-service/txn_lazy_committer.cpp
b/cloud/src/meta-service/txn_lazy_committer.cpp
index 2d209ba2a7f..dfb66e139aa 100644
--- a/cloud/src/meta-service/txn_lazy_committer.cpp
+++ b/cloud/src/meta-service/txn_lazy_committer.cpp
@@ -917,7 +917,7 @@ std::pair<MetaServiceCode, std::string>
TxnLazyCommitTask::commit_partition(
LOG(INFO) << "remove tmp_rowset_key=" << hex(tmp_rowset_key) << "
txn_id=" << txn_id_;
}
- TEST_SYNC_POINT_CALLBACK("TxnLazyCommitter::commit");
+ TEST_SYNC_POINT_CALLBACK("TxnLazyCommitter::commit", &partition_id);
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::COMMIT>(err);
@@ -926,6 +926,7 @@ std::pair<MetaServiceCode, std::string>
TxnLazyCommitTask::commit_partition(
return {code, msg};
}
}
+ TEST_SYNC_POINT_CALLBACK("TxnLazyCommitTask::commit_partition::finish",
&partition_id);
return {MetaServiceCode::OK, ""};
}
diff --git a/cloud/test/txn_lazy_commit_test.cpp
b/cloud/test/txn_lazy_commit_test.cpp
index 7d73e72ac25..edb76aafa12 100644
--- a/cloud/test/txn_lazy_commit_test.cpp
+++ b/cloud/test/txn_lazy_commit_test.cpp
@@ -24,11 +24,14 @@
#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>
@@ -1450,6 +1453,354 @@ TEST(TxnLazyCommitTest,
CommitTxnWithSubTxnRepairTabletIndexTest) {
}
}
+// 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();
@@ -1695,6 +2046,14 @@ TEST(TxnLazyCommitTest,
FallThroughCommitTxnEventuallyTest) {
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);
}
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
