branch-4.0: [fix](transaction) select txn insert backend from current cluster #63634

fe/fe-core/src/main/java/org/apache/doris/qe/InsertStreamTxnExecutor.java

@@ -44,6 +44,7 @@
 
 import org.apache.thrift.TException;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -105,13 +106,7 @@ public void beginTransaction(TStreamLoadPutRequest request) throws UserException
             table.readUnlock();
         }
 
-        BeSelectionPolicy policy = new BeSelectionPolicy.Builder().needLoadAvailable().needQueryAvailable().build();
-        List<Long> beIds = Env.getCurrentSystemInfo().selectBackendIdsByPolicy(policy, 1);
-        if (beIds.isEmpty()) {
-            throw new UserException("No available backend to match the policy: " + policy);
-        }
-
-        Backend backend = Env.getCurrentSystemInfo().getBackendsByCurrentCluster().get(beIds.get(0));
+        Backend backend = selectBackendForTxnLoad();
         txnConf.setUserIp(backend.getHost());
         txnEntry.setBackend(backend);
         TNetworkAddress address = new TNetworkAddress(backend.getHost(), backend.getBrpcPort());
@@ -128,6 +123,18 @@ public void beginTransaction(TStreamLoadPutRequest request) throws UserException
         }
     }
 
+    static Backend selectBackendForTxnLoad() throws UserException {
+        BeSelectionPolicy policy = new BeSelectionPolicy.Builder().needLoadAvailable().needQueryAvailable().build();
+        Map<Long, Backend> currentBackends = Env.getCurrentSystemInfo().getBackendsByCurrentCluster();
+        List<Long> beIds = Env.getCurrentSystemInfo()
+                .selectBackendIdsByPolicy(policy, 1, new ArrayList<>(currentBackends.values()));
+        if (beIds.isEmpty()) {
+            throw new UserException("No available backend to match the policy: " + policy);
+        }
+
+        return currentBackends.get(beIds.get(0));
+    }
+
     public void commitTransaction() throws TException, TimeoutException,
             InterruptedException, ExecutionException {
         TTxnParams txnConf = txnEntry.getTxnConf();

fe/fe-core/src/test/java/org/apache/doris/qe/InsertStreamTxnExecutorTest.java

@@ -0,0 +1,57 @@
+// 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.qe;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class InsertStreamTxnExecutorTest {
+
+    @Test
+    public void testSelectBackendForTxnLoadUsesCurrentClusterBackends() throws Exception {
+        SystemInfoService systemInfoService = Mockito.spy(new SystemInfoService());
+        Backend otherClusterBackend = createBackend(10001L, "127.0.0.1", 9060);
+        Backend currentClusterBackend = createBackend(10002L, "127.0.0.2", 9061);
+        systemInfoService.addBackend(otherClusterBackend);
+        systemInfoService.addBackend(currentClusterBackend);
+        Mockito.doReturn(ImmutableMap.of(currentClusterBackend.getId(), currentClusterBackend))
+                .when(systemInfoService).getBackendsByCurrentCluster();
+
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+            Backend selectedBackend = InsertStreamTxnExecutor.selectBackendForTxnLoad();
+
+            Assert.assertEquals(currentClusterBackend.getId(), selectedBackend.getId());
+        }
+    }
+
+    private Backend createBackend(long id, String host, int brpcPort) {
+        Backend backend = new Backend(id, host, 9050);
+        backend.setAlive(true);
+        backend.setBrpcPort(brpcPort);
+        return backend;
+    }
+}