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 3ef0f866976 branch-4.1: [fix](cloud) Skip BE rowset promotion for
incomplete lazy commits (#66253)
3ef0f866976 is described below
commit 3ef0f866976bcec89f5d542cbebdb019c6c497fe
Author: bobhan1 <[email protected]>
AuthorDate: Mon Aug 3 10:29:38 2026 +0800
branch-4.1: [fix](cloud) Skip BE rowset promotion for incomplete lazy
commits (#66253)
### What problem does this PR solve?
Issue Number: None
Related PR: #66092
Problem Summary:
Backport #66092 to `branch-4.1`.
During a large cloud load, lazy commit can finish its first metadata
phase but fail to complete background rowset promotion before the commit
RPC returns. FE previously treated this response as fully published and
notified BEs to make temporary rowsets visible, which could race the
meta-service metadata update.
This change reports whether the commit used the lazy path and whether
that path is incomplete. FE skips the BE fast-path promotion only for
incomplete lazy commits.
The only cherry-pick conflict was in `CloudGlobalTransactionMgrTest`:
master has the JMockit-to-Mockito migration from #62221, while
`branch-4.1` does not. The conflict was resolved by preserving the
existing `branch-4.1` JMockit tests and adding only the four independent
notification cases and their helper from #66092.
### Release note
Fix a cloud metadata consistency issue between large transaction lazy
commit and BE compaction.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test
- `build-support/check-format.sh`
- `./run-fe-ut.sh --run
org.apache.doris.cloud.transaction.CloudGlobalTransactionMgrTest` (20
tests, 0 failures, 0 errors)
- `./build.sh --cloud -j100`
- `BUILD_TYPE=Release ./run-cloud-ut.sh --run
--filter=txn_lazy_commit_test:TxnLazyCommitTest.CommitTxnEventuallyWithFailedLazyCommitTaskTest
-j100` (1 test passed)
- Full `TxnLazyCommitTest.*`: 22/24 passed in both ASAN and Release.
`CommitTxnEventuallyWithHugeRowsetMetaTest` and
`CommitTxnEventuallyWithManyPartitions` consistently timed out after 20
seconds in MemTxnKv; the new lazy-commit failure test passed in both
runs.
- Behavior changed:
- [ ] No.
- [x] Yes. FE no longer promotes temporary rowsets when meta-service
reports an incomplete lazy commit.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/cloud/cloud_meta_mgr.cpp | 3 +
cloud/src/meta-service/meta_service_txn.cpp | 6 ++
cloud/test/txn_lazy_commit_test.cpp | 111 +++++++++++++++++++++
.../transaction/CloudGlobalTransactionMgr.java | 6 ++
.../transaction/CloudGlobalTransactionMgrTest.java | 77 ++++++++++++++
gensrc/proto/cloud.proto | 5 +
6 files changed, 208 insertions(+)
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
index 909b4a8b5b9..f731186679d 100644
--- a/be/src/cloud/cloud_meta_mgr.cpp
+++ b/be/src/cloud/cloud_meta_mgr.cpp
@@ -1647,6 +1647,9 @@ Status CloudMetaMgr::commit_txn(const StreamLoadContext&
ctx, bool is_2pc) {
});
if (st.ok()) {
+ VLOG_DEBUG << "commit txn succeeded, db_id: " << ctx.db_id << ",
txn_id: " << ctx.txn_id
+ << ", label: " << ctx.label << ", is_lazy_commit: " <<
res.is_lazy_commit()
+ << ", is_lazy_commit_incomplete: " <<
res.is_lazy_commit_incomplete();
std::vector<int64_t> tablet_ids;
for (auto& commit_info : ctx.commit_infos) {
tablet_ids.emplace_back(commit_info.tabletId);
diff --git a/cloud/src/meta-service/meta_service_txn.cpp
b/cloud/src/meta-service/meta_service_txn.cpp
index 5c35958b1f0..33df289d70a 100644
--- a/cloud/src/meta-service/meta_service_txn.cpp
+++ b/cloud/src/meta-service/meta_service_txn.cpp
@@ -2194,6 +2194,7 @@ void MetaServiceImpl::commit_txn_eventually(
std::string& msg, const std::string& instance_id, int64_t db_id,
const std::vector<std::pair<std::string, doris::RowsetMetaCloudPB>>&
tmp_rowsets_meta,
KVStats& stats) {
+ response->set_is_lazy_commit(true);
StopWatch sw;
DORIS_CLOUD_DEFER {
if (config::use_detailed_metrics && !instance_id.empty()) {
@@ -2620,6 +2621,8 @@ void MetaServiceImpl::commit_txn_eventually(
return;
}
+ response->set_is_lazy_commit_incomplete(true);
+
// set table versions in response
if (is_versioned_read) {
Versionstamp vs;
@@ -2649,6 +2652,8 @@ void MetaServiceImpl::commit_txn_eventually(
if (ret.first != MetaServiceCode::OK) {
LOG(WARNING) << "txn lazy commit failed txn_id=" << txn_id << "
code=" << ret.first
<< " msg=" << ret.second;
+ } else {
+ response->set_is_lazy_commit_incomplete(false);
}
std::unordered_map<int64_t, TabletStats> tablet_stats; // tablet_id ->
stats
@@ -3282,6 +3287,7 @@ void
MetaServiceImpl::commit_txn(::google::protobuf::RpcController* controller,
const CommitTxnRequest* request,
CommitTxnResponse* response,
::google::protobuf::Closure* done) {
RPC_PREPROCESS(commit_txn, get, put, del);
+ response->set_is_lazy_commit(false);
if (!request->has_txn_id()) {
code = MetaServiceCode::INVALID_ARGUMENT;
msg = "invalid argument, missing txn id";
diff --git a/cloud/test/txn_lazy_commit_test.cpp
b/cloud/test/txn_lazy_commit_test.cpp
index b1bb8b3a62a..60a33067143 100644
--- a/cloud/test/txn_lazy_commit_test.cpp
+++ b/cloud/test/txn_lazy_commit_test.cpp
@@ -1192,6 +1192,10 @@ TEST(TxnLazyCommitTest, CommitTxnImmediatelyTest) {
&res, nullptr);
ASSERT_EQ(res.status().code(), MetaServiceCode::OK);
ASSERT_TRUE(commit_txn_immediatelly_hit);
+ 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());
}
{
@@ -1207,6 +1211,83 @@ TEST(TxnLazyCommitTest, CommitTxnImmediatelyTest) {
}
}
+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_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;
@@ -2224,6 +2305,10 @@ TEST(TxnLazyCommitTest,
ConcurrentCommitTxnEventuallyCase4Test) {
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());
}
}
@@ -3404,6 +3489,32 @@ TEST(TxnLazyCommitTest,
CommitTxnEventuallyWithAbortAfterCommitTest) {
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());
}
{
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index 3364a83f795..df0e193b0e4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -2781,6 +2781,12 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
*/
private void notifyBesMakeTmpRsVisible(CommitTxnResponse commitTxnResponse,
List<TabletCommitInfo>
tabletCommitInfos) {
+ if (commitTxnResponse.getIsLazyCommit()
+ && commitTxnResponse.getIsLazyCommitIncomplete()) {
+ LOG.info("skip make cloud tmp rowsets visible for incomplete lazy
commit, txn_id: {}",
+ commitTxnResponse.getTxnInfo().getTxnId());
+ return;
+ }
if (tabletCommitInfos == null || tabletCommitInfos.isEmpty()
|| !Config.enable_notify_be_after_load_txn_commit) {
return;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
index 7b900960f22..1daaaf51de8 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
@@ -40,8 +40,10 @@ import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.QuotaExceedException;
import org.apache.doris.common.UserException;
import org.apache.doris.load.routineload.RLTaskTxnCommitAttachment;
+import org.apache.doris.thrift.TTabletCommitInfo;
import org.apache.doris.transaction.BeginTransactionException;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TabletCommitInfo;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TxnStateChangeCallback;
@@ -57,6 +59,10 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
public class CloudGlobalTransactionMgrTest {
@@ -203,6 +209,77 @@ public class CloudGlobalTransactionMgrTest {
transactionId, null, null);
}
+ @Test
+ public void testSkipMakeTmpRsVisibleForIncompleteLazyCommit() throws
Exception {
+ CommitTxnResponse response = CommitTxnResponse.newBuilder()
+ .setTxnInfo(TxnInfoPB.newBuilder().setTxnId(12345L).build())
+ .setIsLazyCommit(true)
+ .setIsLazyCommitIncomplete(true)
+ .build();
+
+ Assert.assertFalse(invokeNotifyBesMakeTmpRsVisible(response));
+ }
+
+ @Test
+ public void testMakeTmpRsVisibleForNonLazyCommitWithIncompleteFlag()
throws Exception {
+ CommitTxnResponse response = CommitTxnResponse.newBuilder()
+ .setTxnInfo(TxnInfoPB.newBuilder().setTxnId(12346L).build())
+ .setIsLazyCommit(false)
+ .setIsLazyCommitIncomplete(true)
+ .build();
+
+ Assert.assertTrue(invokeNotifyBesMakeTmpRsVisible(response));
+ }
+
+ @Test
+ public void testMakeTmpRsVisibleForCompletedLazyCommit() throws Exception {
+ CommitTxnResponse response = CommitTxnResponse.newBuilder()
+ .setTxnInfo(TxnInfoPB.newBuilder().setTxnId(12347L).build())
+ .setIsLazyCommit(true)
+ .setIsLazyCommitIncomplete(false)
+ .build();
+
+ Assert.assertTrue(invokeNotifyBesMakeTmpRsVisible(response));
+ }
+
+ @Test
+ public void testMakeTmpRsVisibleForNonLazyCommit() throws Exception {
+ CommitTxnResponse response = CommitTxnResponse.newBuilder()
+ .setTxnInfo(TxnInfoPB.newBuilder().setTxnId(12348L).build())
+ .setIsLazyCommit(false)
+ .setIsLazyCommitIncomplete(false)
+ .build();
+
+ Assert.assertTrue(invokeNotifyBesMakeTmpRsVisible(response));
+ }
+
+ private boolean invokeNotifyBesMakeTmpRsVisible(CommitTxnResponse
response) throws Exception {
+ boolean originalEnableNotify =
Config.enable_notify_be_after_load_txn_commit;
+ try {
+ Config.enable_notify_be_after_load_txn_commit = true;
+ AtomicBoolean notified = new AtomicBoolean(false);
+ CloudGlobalTransactionMgr transactionMgr = new
CloudGlobalTransactionMgr() {
+ @Override
+ public void sendMakeCloudTmpRsVisibleTasks(long txnId,
+ List<TTabletCommitInfo> commitInfos, Map<Long, Long>
partitionVersionMap,
+ long updateVersionVisibleTime) {
+ notified.set(true);
+ }
+ };
+ Method notifyMethod =
CloudGlobalTransactionMgr.class.getDeclaredMethod(
+ "notifyBesMakeTmpRsVisible", CommitTxnResponse.class,
List.class);
+ notifyMethod.setAccessible(true);
+
+ List<TabletCommitInfo> tabletCommitInfos =
+ Lists.newArrayList(new TabletCommitInfo(10001L, 10002L));
+
+ notifyMethod.invoke(transactionMgr, response, tabletCommitInfos);
+ return notified.get();
+ } finally {
+ Config.enable_notify_be_after_load_txn_commit =
originalEnableNotify;
+ }
+ }
+
@Test
public void testCommitTransactionAlreadyVisible() throws UserException {
new MockUp<MetaServiceProxy>(MetaServiceProxy.class) {
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index e5c458132a9..dcf1e3b28aa 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -1046,6 +1046,11 @@ message CommitTxnResponse {
repeated int64 versions = 5;
repeated TableStatsPB table_stats = 6;
optional int64 version_update_time_ms = 7;
+ // Whether this commit request used the lazy commit path.
+ optional bool is_lazy_commit = 8;
+ // The lazy commit has only completed its first phase. FE must not notify
BE to make
+ // temporary rowsets visible until BE observes the final metadata from
meta-service.
+ optional bool is_lazy_commit_incomplete = 9;
}
message AbortTxnRequest {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
