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 6704c8400b2 branch-4.2: [fix](cloud) Match transaction coordinators by
stable backend ID #68756 (#69345)
6704c8400b2 is described below
commit 6704c8400b2cbefa45dee661a8b23cf42bdd6825
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Oct 21 10:48:54 2026 +0800
branch-4.2: [fix](cloud) Match transaction coordinators by stable backend
ID #68756 (#69345)
Cherry-picked from #68756
Co-authored-by: Tran Ngoc Tuan
<[email protected]>
---
cloud/src/meta-service/meta_service_txn.cpp | 15 ++-
cloud/test/meta_service_test.cpp | 70 +++++++++++++-
.../doris/transaction/GlobalTransactionMgr.java | 4 +-
.../doris/transaction/CloudTxnCoordinatorTest.java | 102 +++++++++++++++++++++
4 files changed, 182 insertions(+), 9 deletions(-)
diff --git a/cloud/src/meta-service/meta_service_txn.cpp
b/cloud/src/meta-service/meta_service_txn.cpp
index b003a5baa14..3e6026ec35a 100644
--- a/cloud/src/meta-service/meta_service_txn.cpp
+++ b/cloud/src/meta-service/meta_service_txn.cpp
@@ -4300,9 +4300,12 @@ void
MetaServiceImpl::abort_txn_with_coordinator(::google::protobuf::RpcControll
return;
}
const auto& coordinate = info_pb.coordinator();
+ // Backend IDs survive address changes; use the address only for
legacy unknown IDs.
+ const bool same_coordinator = (coordinate.id() != 0 &&
request->id() != 0)
+ ? coordinate.id() ==
request->id()
+ : coordinate.ip() ==
request->ip();
if (info_pb.status() == TxnStatusPB::TXN_STATUS_PREPARED &&
- coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
coordinate.id() == request->id() &&
- coordinate.ip() == request->ip() &&
+ coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
same_coordinator &&
coordinate.start_time() < request->start_time()) {
need_commit = true;
TxnInfoPB return_txn_info;
@@ -4404,10 +4407,12 @@ void MetaServiceImpl::get_prepare_txn_by_coordinator(
return TxnErrorCode::TXN_INVALID_DATA;
}
const auto& coordinate = info_pb.coordinator();
+ // Backend IDs survive address changes; use the address only for
legacy unknown IDs.
+ const bool same_coordinator = (coordinate.id() != 0 && request->id()
!= 0)
+ ? coordinate.id() ==
request->id()
+ : coordinate.ip() ==
request->ip();
bool matches = info_pb.status() == TxnStatusPB::TXN_STATUS_PREPARED &&
- coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
- coordinate.ip() == request->ip() &&
- (coordinate.id() == 0 || coordinate.id() ==
request->id());
+ coordinate.sourcetype() == TXN_SOURCE_TYPE_BE &&
same_coordinator;
if (matches && has_start_time_filter) {
matches = coordinate.start_time() < request->start_time();
}
diff --git a/cloud/test/meta_service_test.cpp b/cloud/test/meta_service_test.cpp
index e0d91a4c863..6738b498cc1 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -2922,7 +2922,25 @@ TEST(MetaServiceTest, AbortTxnWithCoordinatorTest) {
AbortTxnWithCoordinatorResponse abort_txn_resp;
abort_txn_req.set_id(coordinator_id);
- abort_txn_req.set_ip(host);
+ abort_txn_req.set_cloud_unique_id(cloud_unique_id);
+ abort_txn_req.set_ip("192.168.100.200:9050"); // Same backend, changed
address.
+
+ // The current backend process must not be aborted even when its address
changed.
+ abort_txn_req.set_start_time(cur_time);
+ meta_service->abort_txn_with_coordinator(&abort_txn_cntl, &abort_txn_req,
&abort_txn_resp,
+ nullptr);
+ ASSERT_EQ(abort_txn_resp.status().code(), MetaServiceCode::OK);
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ std::string value;
+ ASSERT_EQ(txn->get(txn_info_key({mock_instance, db_id, txn_id}),
&value),
+ TxnErrorCode::TXN_OK);
+ TxnInfoPB info;
+ ASSERT_TRUE(info.ParseFromString(value));
+ ASSERT_EQ(info.status(), TxnStatusPB::TXN_STATUS_PREPARED);
+ }
+ abort_txn_resp.Clear();
abort_txn_req.set_start_time(cur_time + 3600);
// first time to check txn conflict
@@ -3062,6 +3080,49 @@ TEST(MetaServiceTest, GetPrepareTxnByCoordinatorTest) {
ASSERT_EQ(resp.status().code(), MetaServiceCode::INVALID_ARGUMENT);
}
+ // A rescheduled backend retains its ID while its address changes. Test
both scan paths.
+ {
+ const bool original_mode =
config::enable_get_prepare_txn_by_coordinator_by_running_key;
+ DORIS_CLOUD_DEFER {
+ config::enable_get_prepare_txn_by_coordinator_by_running_key =
original_mode;
+ };
+ for (bool scan_by_running_key : {false, true}) {
+ config::enable_get_prepare_txn_by_coordinator_by_running_key =
scan_by_running_key;
+ brpc::Controller cntl;
+ GetPrepareTxnByCoordinatorRequest req;
+ GetPrepareTxnByCoordinatorResponse resp;
+ req.set_cloud_unique_id(cloud_unique_id);
+ req.set_id(coordinator_id);
+ req.set_ip("192.168.100.200:9050");
+ meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_EQ(resp.txn_infos_size(), 5);
+
+ // Startup time still protects transactions belonging to the
current process.
+ resp.Clear();
+ req.set_start_time(cur_time);
+ meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_EQ(resp.txn_infos_size(), 0);
+
+ // A reused address with a different known ID is a different
coordinator.
+ resp.Clear();
+ req.clear_start_time();
+ req.set_id(coordinator_id + 1);
+ req.set_ip(host);
+ meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_EQ(resp.txn_infos_size(), 0);
+
+ // An unknown request ID retains address-based compatibility.
+ resp.Clear();
+ req.set_id(0);
+ meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
+ ASSERT_EQ(resp.txn_infos_size(), 5);
+ }
+ }
+
// Running entries also include precommitted/lazy-committed and expired
transactions.
{
std::unique_ptr<Transaction> txn;
@@ -3376,9 +3437,11 @@ TEST(MetaServiceTest,
GetPrepareTxnByCoordinatorScanModeTest) {
req.set_start_time(150);
meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
- ASSERT_EQ(resp.txn_infos_size(), 2);
+ // Txn 3 has the same backend ID despite its changed address.
+ ASSERT_EQ(resp.txn_infos_size(), 3);
EXPECT_EQ(resp.txn_infos(0).txn_id(), 1);
EXPECT_EQ(resp.txn_infos(1).txn_id(), 2);
+ EXPECT_EQ(resp.txn_infos(2).txn_id(), 3);
}
// Skip malformed running keys on the first, middle and last data pages.
@@ -3412,9 +3475,10 @@ TEST(MetaServiceTest,
GetPrepareTxnByCoordinatorScanModeTest) {
req.set_start_time(150);
meta_service->get_prepare_txn_by_coordinator(&cntl, &req, &resp,
nullptr);
ASSERT_EQ(resp.status().code(), MetaServiceCode::OK);
- ASSERT_EQ(resp.txn_infos_size(), 2);
+ ASSERT_EQ(resp.txn_infos_size(), 3);
EXPECT_EQ(resp.txn_infos(0).txn_id(), 1);
EXPECT_EQ(resp.txn_infos(1).txn_id(), 2);
+ EXPECT_EQ(resp.txn_infos(2).txn_id(), 3);
EXPECT_EQ(pages, 9 + 3 + 1);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
index bd1ddf2ba6a..3a0b30da9c8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
@@ -491,8 +491,10 @@ public class GlobalTransactionMgr implements
GlobalTransactionMgrIface {
} else if (coordinator.sourceType ==
TransactionState.TxnSourceType.BE) {
Backend be = Env.getCurrentSystemInfo().getBackend(coordinator.id);
if (be != null) {
+ // Cloud backend registration is stable across pod IP changes
and FQDN addressing.
+ boolean sameInstance = Config.isCloudMode() ||
be.getHost().equals(coordinator.ip);
offline = false;
- if (be.getHost().equals(coordinator.ip) &&
(be.getLastStartTime() > coordinator.startTime
+ if (sameInstance && (be.getLastStartTime() >
coordinator.startTime
|| (!be.isAlive() && System.currentTimeMillis() -
be.getLastUpdateMs()
>=
Config.abort_txn_after_lost_heartbeat_time_second * 1000L))) {
return true;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/transaction/CloudTxnCoordinatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/transaction/CloudTxnCoordinatorTest.java
new file mode 100644
index 00000000000..675500e7f3d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/transaction/CloudTxnCoordinatorTest.java
@@ -0,0 +1,102 @@
+// 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.
+
+package org.apache.doris.transaction;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.transaction.TransactionState.LoadJobSourceType;
+import org.apache.doris.transaction.TransactionState.TxnCoordinator;
+import org.apache.doris.transaction.TransactionState.TxnSourceType;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class CloudTxnCoordinatorTest {
+ private String previousDeployMode;
+ private String previousCloudUniqueId;
+ private MockedStatic<Env> mockedEnv;
+ private Backend backend;
+ private TransactionState transaction;
+
+ @BeforeEach
+ public void setUp() {
+ previousDeployMode = Config.deploy_mode;
+ previousCloudUniqueId = Config.cloud_unique_id;
+ Config.deploy_mode = "cloud";
+ Config.cloud_unique_id = "";
+ backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getHost()).thenReturn("be-0.example");
+ Mockito.when(backend.isAlive()).thenReturn(true);
+ Mockito.when(backend.getLastStartTime()).thenReturn(100L);
+ SystemInfoService systemInfo = Mockito.mock(SystemInfoService.class);
+ Mockito.when(systemInfo.getBackend(123L)).thenReturn(backend);
+ mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfo);
+ transaction = new TransactionState(1L, Lists.newArrayList(2L), 3L,
+ "coordinator", null, LoadJobSourceType.BACKEND_STREAMING,
+ new TxnCoordinator(TxnSourceType.BE, 123L, "old-pod-ip",
100L), -1L, 1000L);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ mockedEnv.close();
+ Config.deploy_mode = previousDeployMode;
+ Config.cloud_unique_id = previousCloudUniqueId;
+ }
+
+ @Test
+ public void testRestartWithDifferentAddress() {
+ Mockito.when(backend.getLastStartTime()).thenReturn(101L);
+
Assertions.assertTrue(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ }
+
+ @Test
+ public void testCurrentProcessWithDifferentAddress() {
+
Assertions.assertFalse(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ }
+
+ @Test
+ public void testLostHeartbeatWithDifferentAddress() {
+ Mockito.when(backend.isAlive()).thenReturn(false);
+ Mockito.when(backend.getLastUpdateMs()).thenReturn(0L);
+
Assertions.assertTrue(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ }
+
+ @Test
+ public void testRecentHeartbeatWithDifferentAddress() {
+ Mockito.when(backend.isAlive()).thenReturn(false);
+
Mockito.when(backend.getLastUpdateMs()).thenReturn(System.currentTimeMillis());
+
Assertions.assertFalse(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ }
+
+ @Test
+ public void testNonCloudRetainsAddressCheck() {
+ Config.deploy_mode = "local";
+ Mockito.when(backend.getLastStartTime()).thenReturn(101L);
+
Assertions.assertFalse(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ Mockito.when(backend.getHost()).thenReturn("old-pod-ip");
+
Assertions.assertTrue(GlobalTransactionMgr.checkFailedTxnsByCoordinator(transaction));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
