[Feature][Connector-V2]Support Doris Fe Node HA

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/rest/RestService.java

@@ -55,9 +55,7 @@
 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;
@@ -240,30 +238,15 @@ static String[] parseIdentifier(String tableIdentifier, Logger logger)
     }
 
     @VisibleForTesting
-    public static String randomEndpoint(String feNodes, Logger logger)
-            throws DorisConnectorException {
-        logger.trace("Parse fenodes '{}'.", feNodes);
-        if (StringUtils.isEmpty(feNodes)) {
-            String errMsg =
-                    String.format(ErrorMessages.ILLEGAL_ARGUMENT_MESSAGE, "fenodes", feNodes);
-            throw new DorisConnectorException(DorisConnectorErrorCode.REST_SERVICE_FAILED, errMsg);
-        }
-        List<String> nodes = Arrays.asList(feNodes.split(","));
-        Collections.shuffle(nodes);
-        return nodes.get(0).trim();
-    }
-
-    @VisibleForTesting
-    static String getUriStr(
-            DorisSourceConfig dorisSourceConfig, DorisSourceTable dorisSourceTable, Logger logger)
+    static String getUriStr(String node, DorisSourceTable dorisSourceTable, Logger logger)
             throws DorisConnectorException {
         String tableIdentifier =
                 dorisSourceTable.getTablePath().getDatabaseName()
                         + "."
                         + dorisSourceTable.getTablePath().getTableName();
         String[] identifier = parseIdentifier(tableIdentifier, logger);
         return "http://"
-                + randomEndpoint(dorisSourceConfig.getFrontends(), logger)
+                + node.trim()
                 + API_PREFIX
                 + "/"
                 + identifier[0]
@@ -298,16 +281,31 @@ public static List<PartitionDefinition> findPartitions(
         }
         logger.debug("Query SQL Sending to Doris FE is: '{}'.", sql);
 
-        HttpPost httpPost =
-                new HttpPost(getUriStr(dorisSourceConfig, dorisSourceTable, logger) + QUERY_PLAN);
         String entity = "{\"sql\": \"" + sql + "\"}";
         logger.debug("Post body Sending to Doris FE is: '{}'.", entity);
         StringEntity stringEntity = new StringEntity(entity, StandardCharsets.UTF_8);
         stringEntity.setContentEncoding("UTF-8");
         stringEntity.setContentType("application/json");
-        httpPost.setEntity(stringEntity);
 
-        String resStr = send(dorisSourceConfig, httpPost, logger);
+        String[] feNodes = dorisSourceConfig.getFrontends().split(",");
+        int feNodesNum = feNodes.length;
+        String resStr = null;
+
+        for (int i = 0; i < feNodesNum; i++) {
+            try {
+                HttpPost httpPost =
+                        new HttpPost(getUriStr(feNodes[i], dorisSourceTable, logger) + QUERY_PLAN);
+                httpPost.setEntity(stringEntity);
+                resStr = send(dorisSourceConfig, httpPost, logger);
+                break;
+            } catch (DorisConnectorException e) {
+                if (i == feNodesNum - 1) {
+                    throw new DorisConnectorException(
+                            DorisConnectorErrorCode.REST_SERVICE_FAILED, e);
+                }
+            }
+        }
+
         logger.debug("Find partition response is '{}'.", resStr);
         QueryPlan queryPlan = getQueryPlan(resStr, logger);
         Map<String, List<Long>> be2Tablets = selectBeForTablet(queryPlan, logger);

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java

@@ -27,7 +27,6 @@
 import org.apache.seatunnel.connectors.doris.config.DorisSinkConfig;
 import org.apache.seatunnel.connectors.doris.exception.DorisConnectorErrorCode;
 import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
-import org.apache.seatunnel.connectors.doris.rest.RestService;
 import org.apache.seatunnel.connectors.doris.rest.models.RespContent;
 import org.apache.seatunnel.connectors.doris.serialize.DorisSerializer;
 import org.apache.seatunnel.connectors.doris.serialize.SeaTunnelRowSerializer;
@@ -98,21 +97,35 @@ public DorisSinkWriter(
     }
 
     private void initializeLoad() {
-        String backend = RestService.randomEndpoint(dorisSinkConfig.getFrontends(), log);
-        try {
-            this.dorisStreamLoad =
-                    new DorisStreamLoad(
-                            backend,
-                            catalogTable.getTablePath(),
-                            dorisSinkConfig,
-                            labelGenerator,
-                            new HttpUtil().getHttpClient());
-            if (dorisSinkConfig.getEnable2PC()) {
-                dorisStreamLoad.abortPreCommit(labelPrefix, lastCheckpointId + 1);
+
+        String[] feNodes = dorisSinkConfig.getFrontends().split(",");
+        int feNodesNum = feNodes.length;
+
+        for (int i = 0; i < feNodesNum; i++) {
+            try {
+                this.dorisStreamLoad =
+                        new DorisStreamLoad(
+                                feNodes[i],
+                                catalogTable.getTablePath(),
+                                dorisSinkConfig,
+                                labelGenerator,
+                                new HttpUtil().getHttpClient());
+                if (dorisSinkConfig.getEnable2PC()) {
+                    dorisStreamLoad.abortPreCommit(labelPrefix, lastCheckpointId + 1);
+                }
+                break;
+            } catch (Exception e) {
+                if (i == feNodesNum - 1) {
+                    throw new DorisConnectorException(
+                            DorisConnectorErrorCode.STREAM_LOAD_FAILED, e);
+                }
+                log.error(
+                        "stream load error for feNode: {} with exception: {}",
+                        feNodes[i],
+                        e.getMessage());
             }
-        } catch (Exception e) {
-            throw new DorisConnectorException(DorisConnectorErrorCode.STREAM_LOAD_FAILED, e);
         }
+
         startLoad(labelGenerator.generateLabel(lastCheckpointId + 1));
         // when uploading data in streaming mode, we need to regularly detect whether there are
         // exceptions.