This is an automated email from the ASF dual-hosted git repository.
yiguolei 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 83f6f9ec033 [fix](cloud) Match transaction coordinators by stable
backend ID (#68756)
83f6f9ec033 is described below
commit 83f6f9ec033ebde4cea139976fb746152c09fd56
Author: Tran Ngoc Tuan <[email protected]>
AuthorDate: Tue Oct 20 16:05:23 2026 +0700
[fix](cloud) Match transaction coordinators by stable backend ID (#68756)
### What problem does this PR solve?
Issue Number: close #68755
Related PR: #65196, #67661, #67761
Problem Summary:
In Apache Doris 4.1.3, a cloud stream-load transaction can remain
PREPARE after its coordinator BE is restarted or rescheduled. With FQDN
addressing, the registered backend hostname differs from the recorded
client IP; after rescheduling, the pod IP can also change. FE failure
detection and meta-service coordinator cleanup then miss the abandoned
transaction, leaving schema changes and rollups in WAITING_TXN until
expiration or manual abort.
This PR targets master and uses the backend ID when both coordinator IDs
are nonzero, falling back to address matching when either ID is unknown.
The cloud FE checks restart/liveness after backend lookup by ID without
requiring hostname/IP equality. Startup-time and transaction-status
filters remain in place, and non-cloud FE address matching is preserved.
Master already contains the schema-change/rollup retry and
committed/final-status safeguards from #65196 and #67661. This PR
complements those changes and preserves the scan/retry behavior from
#67761.
Tests added cover cloud restart and lost-heartbeat detection with
differing addresses, protection of the current process/recent heartbeat,
non-cloud compatibility, changed-address lookup on both meta-service
scan paths, startup-time filtering, different known IDs sharing an
address, unknown request-ID fallback, and changed-address abort with
current-process protection.
Validation:
- On this master-based patch: Java Checkstyle passed for both changed
Java files; changed C++ files were formatted with clang-format 16.0.6;
`git diff --check` passed.
- Source-extracted compiled checks on this patch passed 33 FE
failure-detection cases and 54 identity cases for each of the two
meta-service matching expressions. These checks are isolated harnesses,
not the project's native unit-test suites.
- Earlier Docker validation for the 4.1.3 fix passed
`AlterTxnRecoveryTest` (4 tests) and `CloudIndexTest`,
`SchemaChangeJobV2Test`, `RollupJobV2Test`, and
`GlobalTransactionMgrTest` (37 tests total). These results do not
validate this master PR head.
- The earlier full FE Docker run is still in progress. Two failures have
been observed in `AuthenticationPluginManagerTest` concerning OIDC
factory/plugin loading. The run uses `maven.test.failure.ignore=true` to
collect failures across modules; a successful Maven exit must not be
interpreted as a passing suite.
- Native execution of the new master-based FE and cloud tests,
full-suite validation of this PR head, and a cloud BE-rescheduling
end-to-end reproduction are pending. No successful result is claimed for
them.
CI / review follow-up: request CI workflow approval from a Committer,
then trigger `run buildall` for the current PR head according to the
project's review-and-CI guide. Cloud/meta-service Code Owner review is
required. This PR is ready for code review; pending validation and
required approvals must be resolved before merge.
### Release note
Fix cloud transaction cleanup missing restarted or rescheduled BE
coordinators when their address changes or FQDN addressing is used,
which could leave schema changes and rollups waiting for abandoned
prepared transactions.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Cloud cleanup matches known backend IDs despite address
changes; unknown IDs fall back to address matching.
- 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
---
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 a766482d13e..caa967f2ee7 100644
--- a/cloud/src/meta-service/meta_service_txn.cpp
+++ b/cloud/src/meta-service/meta_service_txn.cpp
@@ -4700,9 +4700,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;
@@ -4804,10 +4807,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 a2ec86680de..c7cdf16e175 100644
--- a/cloud/test/meta_service_test.cpp
+++ b/cloud/test/meta_service_test.cpp
@@ -3014,7 +3014,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
@@ -3154,6 +3172,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;
@@ -3468,9 +3529,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.
@@ -3504,9 +3567,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 ae6a1abcd36..295797dd204 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
@@ -501,8 +501,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]
