Support IPv6 endpoint URLs

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java

@@ -44,6 +44,7 @@ public class ClusterConstant {
   public static final String DEFAULT_DATA_NODE_COMMON_PROPERTIES =
       "DefaultDataNodeCommonProperties";
   public static final String DATA_REGION_PER_DATANODE = "integrationTest.dataRegionPerDataNode";
+  public static final String TEST_NODE_ADDRESS = "IoTDBTestNodeAddress";
 
   // Cluster Configurations
   public static final String CLUSTER_CONFIGURATIONS = "ClusterConfigurations";

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java

@@ -64,6 +64,7 @@
 import org.apache.iotdb.jdbc.IoTDBConnection;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
 import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.UrlUtils;
 import org.apache.iotdb.session.Session;
 import org.apache.iotdb.session.TableSessionBuilder;
 import org.apache.iotdb.session.pool.SessionPool;
@@ -151,9 +152,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
       final String configNodeMetricContent =
           getUrlContent(
               Config.IOTDB_HTTP_URL_PREFIX
-                  + configNode.getIp()
-                  + ":"
-                  + configNode.getMetricPort()
+                  + UrlUtils.formatTEndPointIpv4AndIpv6Url(
+                      configNode.getIp(), configNode.getMetricPort())
                   + "/metrics",
               authHeader);
       result.add(configNodeMetricContent);
@@ -163,9 +163,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
       final String dataNodeMetricContent =
           getUrlContent(
               Config.IOTDB_HTTP_URL_PREFIX
-                  + dataNode.getIp()
-                  + ":"
-                  + dataNode.getMetricPort()
+                  + UrlUtils.formatTEndPointIpv4AndIpv6Url(
+                      dataNode.getIp(), dataNode.getMetricPort())
                   + "/metrics",
               authHeader);
       result.add(dataNodeMetricContent);
@@ -1042,7 +1041,8 @@ protected NodeConnection getWriteConnectionWithSpecifiedDataNode(
       final String password,
       final String sqlDialect)
       throws SQLException {
-    final String endpoint = dataNode.getIp() + ":" + dataNode.getPort();
+    final String endpoint =
+        UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(), dataNode.getPort());
     final Connection writeConnection =
         DriverManager.getConnection(
             Config.IOTDB_URL_PREFIX
@@ -1633,18 +1633,14 @@ public void ensureNodeStatus(
             .forEach(
                 node ->
                     nodeIds.put(
-                        node.getInternalEndPoint().getIp()
-                            + ":"
-                            + node.getInternalEndPoint().getPort(),
+                        UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getInternalEndPoint()),
                         node.getConfigNodeId()));
         showClusterResp
             .getDataNodeList()
             .forEach(
                 node ->
                     nodeIds.put(
-                        node.getClientRpcEndPoint().getIp()
-                            + ":"
-                            + node.getClientRpcEndPoint().getPort(),
+                        UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getClientRpcEndPoint()),
                         node.getDataNodeId()));
         for (int j = 0; j < nodes.size(); j++) {
           BaseNodeWrapper nodeWrapper = nodes.get(j);
@@ -1667,10 +1663,9 @@ public void ensureNodeStatus(
             continue;
           }
           if (nodeWrapper instanceof DataNodeWrapper && targetStatus.equals(NodeStatus.Running)) {
-            final String[] ipPort = nodeWrapper.getIpAndPortString().split(":");
-            final String ip = ipPort[0];
-            final int port = Integer.parseInt(ipPort[1]);
-            try (TSocket socket = new TSocket(new TConfiguration(), ip, port, 1000)) {
+            try (TSocket socket =
+                new TSocket(
+                    new TConfiguration(), nodeWrapper.getIp(), nodeWrapper.getPort(), 1000)) {
               socket.open();
             } catch (final TTransportException e) {
               errorMessages.add(

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java

@@ -27,6 +27,7 @@
 import org.apache.iotdb.it.env.cluster.config.MppJVMConfig;
 import org.apache.iotdb.it.framework.IoTDBTestLogger;
 import org.apache.iotdb.itbase.env.BaseNodeWrapper;
+import org.apache.iotdb.rpc.UrlUtils;
 
 import org.apache.tsfile.external.commons.io.FileUtils;
 import org.apache.tsfile.external.commons.io.file.PathUtils;
@@ -123,6 +124,7 @@
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.TARGET;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_LIB_PATH;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_PATH;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEST_NODE_ADDRESS;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.TRIGGER_LIB_DIR;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.UDF_LIB_DIR;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.USER_DIR;
@@ -181,7 +183,7 @@ protected AbstractNodeWrapper(
     this.testClassName = testClassName;
     this.testMethodName = testMethodName;
     this.portList = portList;
-    this.nodeAddress = "127.0.0.1";
+    this.nodeAddress = System.getProperty(TEST_NODE_ADDRESS, "127.0.0.1");
     this.nodePort = portList[0];
     this.metricPort = portList[portList.length - 2];
     jmxPort = this.portList[portList.length - 1];
@@ -606,7 +608,7 @@ public void setPort(int port) {
 
   @Override
   public final String getIpAndPortString() {
-    return this.getIp() + ":" + this.getPort();
+    return UrlUtils.formatTEndPointIpv4AndIpv6Url(this.getIp(), this.getPort());
   }
 
   protected String workDirFilePath(String dirName, String fileName) {

integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java

@@ -42,6 +42,7 @@
 import org.apache.iotdb.jdbc.Config;
 import org.apache.iotdb.jdbc.Constant;
 import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.UrlUtils;
 import org.apache.iotdb.session.Session;
 import org.apache.iotdb.session.TableSessionBuilder;
 import org.apache.iotdb.session.pool.SessionPool;
@@ -72,6 +73,10 @@ public class RemoteServerEnv implements BaseEnv {
   private IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager;
   private RemoteClusterConfig clusterConfig = new RemoteClusterConfig();
 
+  private String remoteEndpointUrl(String port) {
+    return UrlUtils.formatTEndPointIpv4AndIpv6Url(ip_addr, Integer.parseInt(port));
+  }
+
   @Override
   public void initClusterEnvironment() {
     try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -116,11 +121,11 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
     List<String> result = new ArrayList<>();
     result.add(
         getUrlContent(
-            Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + configNodeMetricPort + "/metrics",
+            Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(configNodeMetricPort) + "/metrics",
             authHeader));
     result.add(
         getUrlContent(
-            Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + dataNodeMetricPort + "/metrics",
+            Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(dataNodeMetricPort) + "/metrics",
             authHeader));
     return result;
   }
@@ -133,7 +138,7 @@ public Connection getConnection(String username, String password, String sqlDial
       Class.forName(Config.JDBC_DRIVER_NAME);
       connection =
           DriverManager.getConnection(
-              Config.IOTDB_URL_PREFIX + ip_addr + ":" + port,
+              Config.IOTDB_URL_PREFIX + remoteEndpointUrl(port),
               BaseEnv.constructProperties(username, password, sqlDialect));
     } catch (ClassNotFoundException e) {
       e.printStackTrace();
@@ -170,9 +175,7 @@ public Connection getConnection(
       connection =
           DriverManager.getConnection(
               Config.IOTDB_URL_PREFIX
-                  + ip_addr
-                  + ":"
-                  + port
+                  + remoteEndpointUrl(port)
                   + "?"
                   + VERSION
                   + "="
@@ -300,15 +303,15 @@ public ISession getSessionConnection(ZoneId zoneId) throws IoTDBConnectionExcept
   @Override
   public ITableSession getTableSessionConnection() throws IoTDBConnectionException {
     return new TableSessionBuilder()
-        .nodeUrls(Collections.singletonList(ip_addr + ":" + port))
+        .nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
         .build();
   }
 
   @Override
   public ITableSession getTableSessionConnectionWithDB(String database)
       throws IoTDBConnectionException {
     return new TableSessionBuilder()
-        .nodeUrls(Collections.singletonList(ip_addr + ":" + port))
+        .nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
         .database(database)
         .build();
   }
@@ -332,7 +335,7 @@ public ITableSession getTableSessionConnection(List<String> nodeUrls)
   public ITableSession getTableSessionConnection(String userName, String password)
       throws IoTDBConnectionException {
     return new TableSessionBuilder()
-        .nodeUrls(Collections.singletonList(ip_addr + ":" + port))
+        .nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
         .username(userName)
         .password(password)
         .build();
@@ -356,7 +359,7 @@ public ISession getSessionConnection(String userName, String password)
   public ISession getSessionConnection(List<String> nodeUrls) throws IoTDBConnectionException {
     Session session =
         new Session.Builder()
-            .nodeUrls(Collections.singletonList(ip_addr + ":" + port))
+            .nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
             .username(SessionConfig.DEFAULT_USER)
             .password(SessionConfig.DEFAULT_PASSWORD)
             .fetchSize(SessionConfig.DEFAULT_FETCH_SIZE)

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java

@@ -0,0 +1,124 @@
+/*
+ * 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.iotdb.db.it;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.it.utils.IPv6TestUtils;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({ClusterIT.class})
+public class IoTDBIPv6ClusterIT {
+
+  private static String previousTestNodeAddress;
+
+  @BeforeClass
+  public static void setUp() {
+    IPv6TestUtils.assumeIPv6LoopbackAvailable();
+    previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
+    EnvFactory.getEnv()
+        .getConfig()
+        .getCommonConfig()
+        .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+        .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+        .setSchemaReplicationFactor(3)
+        .setDataReplicationFactor(3);
+    EnvFactory.getEnv().initClusterEnvironment(1, 3);
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+    IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
+  }
+
+  @Test
+  public void clusterCanCommunicateThroughIPv6Loopback() throws Exception {
+    for (ConfigNodeWrapper configNode : EnvFactory.getEnv().getConfigNodeWrapperList()) {
+      assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getIp());
+      assertTrue(configNode.getIpAndPortString().startsWith("[::1]:"));
+    }
+    for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) {
+      assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
+      assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:"));
+    }
+
+    try (SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+      final TShowClusterResp showClusterResp = client.showCluster();
+      final TSStatus status = showClusterResp.getStatus();
+      assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+      assertEquals(1, showClusterResp.getConfigNodeListSize());
+      assertEquals(3, showClusterResp.getDataNodeListSize());
+      showClusterResp
+          .getConfigNodeList()
+          .forEach(
+              configNode ->
+                  assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getInternalEndPoint().getIp()));
+      showClusterResp.getDataNodeList().forEach(this::assertDataNodeLocationUsesIPv6);
+    }
+
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("CREATE DATABASE root.ipv6_cluster");
+      statement.execute("CREATE TIMESERIES root.ipv6_cluster.d1.s1 INT64");
+      statement.execute("INSERT INTO root.ipv6_cluster.d1(time, s1) VALUES (1, 300)");
+      try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.ipv6_cluster.d1")) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertEquals(300L, resultSet.getLong(2));
+        assertFalse(resultSet.next());
+      }
+    }
+  }
+
+  private void assertDataNodeLocationUsesIPv6(final TDataNodeLocation dataNodeLocation) {
+    assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getClientRpcEndPoint().getIp());
+    assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getInternalEndPoint().getIp());
+    assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getMPPDataExchangeEndPoint().getIp());
+    assertEquals(
+        IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getSchemaRegionConsensusEndPoint().getIp());
+    assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getDataRegionConsensusEndPoint().getIp());
+  }
+}

integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.UrlUtils;
 
 import org.apache.tsfile.utils.Pair;
 import org.awaitility.Awaitility;
@@ -283,7 +284,9 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
       LOGGER.info("Step 6: Verifying schema consistency on each DataNode independently...");
       List<DataNodeWrapper> dataNodeWrappers = EnvFactory.getEnv().getDataNodeWrapperList();
       for (DataNodeWrapper wrapper : dataNodeWrappers) {
-        String nodeDescription = "DataNode " + wrapper.getIp() + ":" + wrapper.getPort();
+        String nodeDescription =
+            "DataNode "
+                + UrlUtils.formatTEndPointIpv4AndIpv6Url(wrapper.getIp(), wrapper.getPort());
         LOGGER.info("Verifying schema on {}", nodeDescription);
         Awaitility.await()
             .atMost(60, TimeUnit.SECONDS)
@@ -307,7 +310,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
       LOGGER.info(
           "Step 7: Stopping each DataNode in turn and verifying remaining nodes show consistent schema...");
       for (DataNodeWrapper stoppedNode : dataNodeWrappers) {
-        String stoppedDesc = "DataNode " + stoppedNode.getIp() + ":" + stoppedNode.getPort();
+        String stoppedDesc =
+            "DataNode "
+                + UrlUtils.formatTEndPointIpv4AndIpv6Url(
+                    stoppedNode.getIp(), stoppedNode.getPort());
         LOGGER.info("Stopping {}", stoppedDesc);
         stoppedNode.stopForcibly();
         Assert.assertFalse(stoppedDesc + " should be stopped", stoppedNode.isAlive());
@@ -318,7 +324,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
             if (aliveNode == stoppedNode) {
               continue;
             }
-            String aliveDesc = "DataNode " + aliveNode.getIp() + ":" + aliveNode.getPort();
+            String aliveDesc =
+                "DataNode "
+                    + UrlUtils.formatTEndPointIpv4AndIpv6Url(
+                        aliveNode.getIp(), aliveNode.getPort());
             Awaitility.await()
                 .pollDelay(1, TimeUnit.SECONDS)
                 .atMost(90, TimeUnit.SECONDS)
@@ -446,7 +455,9 @@ private void verifyTimeSeriesAfterDelete(Statement statement, String context) th
   protected void waitForReplicationComplete(DataNodeWrapper leaderNode) {
     final long timeoutSeconds = 120;
     final String metricsUrl =
-        "http://" + leaderNode.getIp() + ":" + leaderNode.getMetricPort() + "/metrics";
+        "http://"
+            + UrlUtils.formatTEndPointIpv4AndIpv6Url(leaderNode.getIp(), leaderNode.getMetricPort())
+            + "/metrics";
     LOGGER.info(
         "Waiting for consensus pipe syncLag to reach 0 on leader DataNode (url: {}, timeout: {}s)...",
         metricsUrl,