flink-sink support nodeurls to multi ip

connectors/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java

@@ -77,8 +77,15 @@ public void open(Configuration parameters) throws Exception {
   }
 
   void initSession() {
-    pool =
-        new SessionPool(
+    if (options.getNodeUrls() != null) {
+      pool = new SessionPool(
+              options.getNodeUrls(),
+              options.getUser(),
+              options.getPassword(),
+              sessionPoolSize);
+      return;
+    }
+    pool = new SessionPool(
             options.getHost(),
             options.getPort(),
             options.getUser(),

connectors/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/options/IoTDBOptions.java

@@ -19,13 +19,15 @@
 package org.apache.iotdb.flink.options;
 
 import java.io.Serializable;
+import java.util.List;
 
 public class IoTDBOptions implements Serializable {
 
   protected String host;
   protected int port;
   protected String user;
   protected String password;
+  protected List<String> nodeUrls;
 
   public IoTDBOptions(String host, int port, String user, String password) {
     this.host = host;
@@ -34,6 +36,12 @@ public IoTDBOptions(String host, int port, String user, String password) {
     this.password = password;
   }
 
+  public IoTDBOptions(List<String> nodeUrls, String user, String password) {
+    this.nodeUrls = nodeUrls;
+    this.user = user;
+    this.password = password;
+  }
+
   public IoTDBOptions() {}
 
   public String getHost() {
@@ -67,4 +75,12 @@ public String getPassword() {
   public void setPassword(String password) {
     this.password = password;
   }
+
+  public List<String> getNodeUrls() {
+    return nodeUrls;
+  }
+
+  public void setNodeUrls(List<String> nodeUrls) {
+    this.nodeUrls = nodeUrls;
+  }
 }

connectors/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/options/IoTDBSinkOptions.java

@@ -42,6 +42,15 @@ public IoTDBSinkOptions(
     this.timeseriesOptionList = timeseriesOptionList;
   }
 
+  public IoTDBSinkOptions(
+          List<String> nodeUrls,
+          String user,
+          String password,
+          List<TimeseriesOption> timeseriesOptionList) {
+    super(nodeUrls, user, password);
+    this.timeseriesOptionList = timeseriesOptionList;
+  }
+
   public List<TimeseriesOption> getTimeseriesOptionList() {
     return timeseriesOptionList;
   }

connectors/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchInsertTest.java

@@ -25,10 +25,12 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -113,4 +115,13 @@ public void close() throws Exception {
             any(List.class), any(List.class), any(List.class), any(List.class), any(List.class));
     verify(pool).close();
   }
+
+  @Test
+  public void testInitUrlNodes() {
+    List<String> nodeUrls = new ArrayList<>();
+    nodeUrls.add("127.0.0.1:6667");
+    nodeUrls.add("127.0.0.2:6667");
+    IoTDBSinkOptions sinkOptions = new IoTDBSinkOptions(nodeUrls, null, null, null);
+    assertEquals(sinkOptions.getNodeUrls(), nodeUrls);
+  }
 }