random node

TongchengOpenSource · Dec 17, 2024 · 8be7032 · 8be7032
1 parent 17aac54
commit 8be7032
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 7 deletions.
diff --git a/...connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/rest/RestService.java b/...connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/rest/RestService.java
@@ -55,7 +55,9 @@
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -287,14 +289,16 @@ public static List<PartitionDefinition> findPartitions(
         stringEntity.setContentEncoding("UTF-8");
         stringEntity.setContentType("application/json");
 
-        String[] feNodes = dorisSourceConfig.getFrontends().split(",");
-        int feNodesNum = feNodes.length;
+        List<String> feNodes = Arrays.asList(dorisSourceConfig.getFrontends().split(","));
+        Collections.shuffle(feNodes);
+        int feNodesNum = feNodes.size();
         String resStr = null;
 
         for (int i = 0; i < feNodesNum; i++) {
             try {
                 HttpPost httpPost =
-                        new HttpPost(getUriStr(feNodes[i], dorisSourceTable, logger) + QUERY_PLAN);
+                        new HttpPost(
+                                getUriStr(feNodes.get(i), dorisSourceTable, logger) + QUERY_PLAN);
                 httpPost.setEntity(stringEntity);
                 resStr = send(dorisSourceConfig, httpPost, logger);
                 break;

diff --git a/...oris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java b/...oris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
@@ -98,14 +98,15 @@ public DorisSinkWriter(
 
     private void initializeLoad() {
 
-        String[] feNodes = dorisSinkConfig.getFrontends().split(",");
-        int feNodesNum = feNodes.length;
+        List<String> feNodes = Arrays.asList(dorisSinkConfig.getFrontends().split(","));
+        Collections.shuffle(feNodes);
+        int feNodesNum = feNodes.size();
 
         for (int i = 0; i < feNodesNum; i++) {
             try {
                 this.dorisStreamLoad =
                         new DorisStreamLoad(
-                                feNodes[i],
+                                feNodes.get(i),
                                 catalogTable.getTablePath(),
                                 dorisSinkConfig,
                                 labelGenerator,
@@ -121,7 +122,7 @@ private void initializeLoad() {
                 }
                 log.error(
                         "stream load error for feNode: {} with exception: {}",
-                        feNodes[i],
+                        feNodes.get(i),
                         e.getMessage());
             }
         }