This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new e8d22e4b226 branch-4.1: [fix](transaction) avoid aborting newer
same-label transactions #64939 (#65275)
e8d22e4b226 is described below
commit e8d22e4b2269929d541f5dbb167487dba508450c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jul 7 13:42:29 2026 +0800
branch-4.1: [fix](transaction) avoid aborting newer same-label transactions
#64939 (#65275)
Cherry-picked from #64939
Co-authored-by: hui lai <[email protected]>
---
be/src/cloud/cloud_meta_mgr.cpp | 7 ++--
be/test/cloud/cloud_meta_mgr_test.cpp | 62 +++++++++++++++++++++++++++++++++++
2 files changed, 66 insertions(+), 3 deletions(-)
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 01602233b9c..13ff0860d86 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -1668,16 +1668,17 @@ Status CloudMetaMgr::abort_txn(const StreamLoadContext&
ctx) {
AbortTxnResponse res;
req.set_cloud_unique_id(config::cloud_unique_id);
req.set_reason(std::string(ctx.status.msg().substr(0, 1024)));
- if (ctx.db_id > 0 && !ctx.label.empty()) {
+ if (ctx.txn_id > 0) {
+ req.set_txn_id(ctx.txn_id);
+ } else if (ctx.db_id > 0 && !ctx.label.empty()) {
req.set_db_id(ctx.db_id);
req.set_label(ctx.label);
- } else if (ctx.txn_id > 0) {
- req.set_txn_id(ctx.txn_id);
} else {
LOG(WARNING) << "failed abort txn, with illegal input, db_id=" <<
ctx.db_id
<< " txn_id=" << ctx.txn_id << " label=" << ctx.label;
return Status::InternalError<false>("failed to abort txn");
}
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudMetaMgr::abort_txn.before_rpc",
Status::OK(), &req);
return retry_rpc(MetaServiceRPC::ABORT_TXN, req, &res,
&MetaService_Stub::abort_txn,
{
.host_limiters = host_level_ms_rpc_rate_limiters_,
diff --git a/be/test/cloud/cloud_meta_mgr_test.cpp
b/be/test/cloud/cloud_meta_mgr_test.cpp
index ff4539f5969..509bdff0428 100644
--- a/be/test/cloud/cloud_meta_mgr_test.cpp
+++ b/be/test/cloud/cloud_meta_mgr_test.cpp
@@ -28,6 +28,7 @@
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cpp/sync_point.h"
+#include "load/stream_load/stream_load_context.h"
#include "storage/olap_common.h"
#include "storage/rowset/rowset_factory.h"
#include "storage/rowset/rowset_meta.h"
@@ -43,6 +44,35 @@ class CloudMetaMgrTest : public testing::Test {
void TearDown() override {}
};
+static AbortTxnRequest get_abort_txn_request(CloudMetaMgr* meta_mgr, const
StreamLoadContext& ctx) {
+ auto* sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ bool called = false;
+ AbortTxnRequest captured_req;
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ "CloudMetaMgr::abort_txn.before_rpc",
+ [&](auto&& args) {
+ auto* req = try_any_cast<AbortTxnRequest*>(args[0]);
+ captured_req.CopyFrom(*req);
+ called = true;
+ auto* ret = try_any_cast<std::pair<Status,
bool>*>(args.back());
+ ret->first = Status::OK();
+ ret->second = true;
+ },
+ &guard);
+
+ Status status = meta_mgr->abort_txn(ctx);
+
+ EXPECT_TRUE(status.ok()) << "Status: " << status;
+ EXPECT_TRUE(called);
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ return captured_req;
+}
+
TEST_F(CloudMetaMgrTest, bthread_fork_join_test) {
// clang-format off
std::atomic<int> task_counter{0};
@@ -170,6 +200,38 @@ TEST_F(CloudMetaMgrTest, bthread_fork_join_test) {
// clang-format on
}
+TEST_F(CloudMetaMgrTest, abort_txn_prefers_txn_id_when_label_is_also_present) {
+ CloudMetaMgr meta_mgr;
+ StreamLoadContext ctx(nullptr);
+ ctx.db_id = 10001;
+ ctx.txn_id = 20002;
+ ctx.label = "same_label";
+ ctx.status = Status::InternalError<false>("load failed");
+
+ AbortTxnRequest captured_req = get_abort_txn_request(&meta_mgr, ctx);
+
+ EXPECT_TRUE(captured_req.has_txn_id());
+ EXPECT_EQ(captured_req.txn_id(), ctx.txn_id);
+ EXPECT_FALSE(captured_req.has_db_id());
+ EXPECT_FALSE(captured_req.has_label());
+}
+
+TEST_F(CloudMetaMgrTest, abort_txn_uses_label_when_txn_id_is_missing) {
+ CloudMetaMgr meta_mgr;
+ StreamLoadContext ctx(nullptr);
+ ctx.db_id = 10001;
+ ctx.label = "same_label";
+ ctx.status = Status::InternalError<false>("load failed");
+
+ AbortTxnRequest captured_req = get_abort_txn_request(&meta_mgr, ctx);
+
+ EXPECT_FALSE(captured_req.has_txn_id());
+ EXPECT_TRUE(captured_req.has_db_id());
+ EXPECT_EQ(captured_req.db_id(), ctx.db_id);
+ EXPECT_TRUE(captured_req.has_label());
+ EXPECT_EQ(captured_req.label(), ctx.label);
+}
+
TEST_F(CloudMetaMgrTest, test_fill_version_holes_no_holes) {
CloudStorageEngine engine(EngineOptions {});
CloudMetaMgr meta_mgr;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
