This is an automated email from the ASF dual-hosted git repository.
sollhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 49af19213e7 [fix](transaction) avoid aborting newer same-label
transactions (#64939)
49af19213e7 is described below
commit 49af19213e735a021661e6aa03d0b21e1c25090f
Author: hui lai <[email protected]>
AuthorDate: Mon Jul 6 20:41:30 2026 +0800
[fix](transaction) avoid aborting newer same-label transactions (#64939)
### What problem does this PR solve?
In cloud mode, `CloudMetaMgr::abort_txn` currently prefers `db_id +
label` over `txn_id` when building `AbortTxnRequest`.
For stream load rollback, the context usually contains both `txn_id` and
`label`. If an old stream load attempt is aborted externally but its
BE-side execution is still cleaning up later, its rollback may abort by
label instead of by the original txn id.
When the upstream retries with the same label, this label-based rollback
can match and abort a newer running transaction with the same label,
causing the new attempt to fail with:
```
txn is not in 1 state, txn_status=4
```
### Solution
Prefer txn_id when aborting a cloud stream load transaction. Fall back
to db_id + label only when txn_id is not available.
This makes rollback cleanup target the exact failed transaction, instead
of affecting another transaction that happens to use the same label.
---
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 479e9fb153d..a582f83991e 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -1669,16 +1669,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 a6149cae9c5..c6e99cb26b1 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]
