This is an automated email from the ASF dual-hosted git repository.
HTHou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 0ef17ba7677 Support IPv6 endpoint URLs (#18162)
0ef17ba7677 is described below
commit 0ef17ba76771bf34d9c3fe6f6041a068d7dea7a4
Author: Haonan <[email protected]>
AuthorDate: Wed Jul 15 19:04:59 2026 +0800
Support IPv6 endpoint URLs (#18162)
Co-authored-by: Caideyipi <[email protected]>
---
.../iotdb/it/env/cluster/ClusterConstant.java | 1 +
.../iotdb/it/env/cluster/env/AbstractEnv.java | 29 ++-
.../it/env/cluster/node/AbstractNodeWrapper.java | 6 +-
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 23 ++-
.../org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java | 124 ++++++++++++
.../iotdb/db/it/IoTDBIPv6ExternalServiceIT.java | 220 +++++++++++++++++++++
.../IoTDBIoTConsensusV23C3DBasicITBase.java | 19 +-
.../org/apache/iotdb/it/utils/IPv6TestUtils.java | 60 ++++++
.../manual/enhanced/IoTDBPipeClusterIT.java | 2 +-
.../dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java | 166 ++++++++++++++++
.../auto/enhanced/IoTDBPipeClusterIT.java | 2 +-
.../apache/iotdb/session/it/IoTDBIPv6ClientIT.java | 142 +++++++++++++
iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp | 5 +-
iotdb-client/client-cpp/src/rpc/RpcCommon.cpp | 23 +++
iotdb-client/client-cpp/src/rpc/RpcCommon.h | 2 +
iotdb-client/client-cpp/src/session/Session.cpp | 4 +-
iotdb-client/client-cpp/test/cpp/sessionIT.cpp | 15 ++
iotdb-client/client-cpp/test/main.cpp | 5 +-
iotdb-client/client-py/iotdb/Session.py | 85 ++++++--
.../client-py/tests/unit/test_session_node_url.py | 85 ++++++++
.../src/main/java/org/apache/iotdb/jdbc/Utils.java | 77 +++++---
.../test/java/org/apache/iotdb/jdbc/UtilsTest.java | 46 +++++
.../main/java/org/apache/iotdb/rpc/UrlUtils.java | 85 ++++++--
.../java/org/apache/iotdb/rpc/UrlUtilsTest.java | 80 ++++++++
.../apache/iotdb/session/i18n/SessionMessages.java | 2 +
.../apache/iotdb/session/i18n/SessionMessages.java | 2 +
.../org/apache/iotdb/session/NodesSupplier.java | 5 +-
.../java/org/apache/iotdb/session/Session.java | 5 +-
.../apache/iotdb/session/SessionConnection.java | 11 +-
.../org/apache/iotdb/session/pool/SessionPool.java | 7 +-
.../apache/iotdb/session/util/SessionUtils.java | 8 +-
iotdb-core/ainode/iotdb/ainode/core/config.py | 17 +-
.../ainode/iotdb/ainode/core/ingress/iotdb.py | 11 +-
.../ainode/resources/conf/iotdb-ainode.properties | 3 +-
iotdb-core/ainode/tests/test_config.py | 71 +++++++
.../apache/iotdb/consensus/ratis/utils/Utils.java | 9 +-
.../iotdb/consensus/ratis/utils/UtilsTest.java | 54 +++++
.../client/IoTDBDataNodeAsyncClientManager.java | 8 +-
.../client/IoTDBDataNodeSyncClientManager.java | 6 +-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 3 +-
.../execution/config/sys/TestConnectionTask.java | 3 +-
.../sink/IoTDBDataNodeAsyncClientManagerTest.java | 120 +++++++++++
.../conf/iotdb-system.properties.template | 10 +-
.../apache/iotdb/commons/utils/NodeUrlUtils.java | 9 +-
.../iotdb/commons/utils/NodeUrlUtilsTest.java | 7 +
45 files changed, 1532 insertions(+), 145 deletions(-)
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java
index 691a13509d9..9118f51f436 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java
+++
b/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";
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
index ae724c7081d..3819bd32a8d 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -64,6 +64,7 @@ import org.apache.iotdb.jdbc.Constant;
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 abstract class AbstractEnv implements BaseEnv {
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 abstract class AbstractEnv implements BaseEnv {
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 @@ public abstract class AbstractEnv implements BaseEnv {
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 abstract class AbstractEnv implements BaseEnv {
.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 abstract class AbstractEnv implements BaseEnv {
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(
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
index 5a271c1ead2..8d3f5f2cc9d 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
+++
b/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.MppCommonConfig;
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.TAB;
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 @@ public abstract class AbstractNodeWrapper implements
BaseNodeWrapper {
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 abstract class AbstractNodeWrapper implements
BaseNodeWrapper {
@Override
public final String getIpAndPortString() {
- return this.getIp() + ":" + this.getPort();
+ return UrlUtils.formatTEndPointIpv4AndIpv6Url(this.getIp(),
this.getPort());
}
protected String workDirFilePath(String dirName, String fileName) {
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
index 586eff60494..3790c37dcea 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -42,6 +42,7 @@ import org.apache.iotdb.itbase.env.ClusterConfig;
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 class RemoteServerEnv implements BaseEnv {
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 class RemoteServerEnv implements BaseEnv {
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 class RemoteServerEnv implements BaseEnv {
connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX
- + ip_addr
- + ":"
- + port
+ + remoteEndpointUrl(port)
+ "?"
+ VERSION
+ "="
@@ -300,7 +303,7 @@ public class RemoteServerEnv implements BaseEnv {
@Override
public ITableSession getTableSessionConnection() throws
IoTDBConnectionException {
return new TableSessionBuilder()
- .nodeUrls(Collections.singletonList(ip_addr + ":" + port))
+ .nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
.build();
}
@@ -308,7 +311,7 @@ public class RemoteServerEnv implements BaseEnv {
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 class RemoteServerEnv implements BaseEnv {
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 class RemoteServerEnv implements BaseEnv {
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)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java
new file mode 100644
index 00000000000..c01e89c3b98
--- /dev/null
+++
b/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());
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ExternalServiceIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ExternalServiceIT.java
new file mode 100644
index 00000000000..e63d428992c
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ExternalServiceIT.java
@@ -0,0 +1,220 @@
+/*
+ * 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.isession.ISession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+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.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.UrlUtils;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.util.EntityUtils;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.RowRecord;
+import org.awaitility.Awaitility;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.QoS;
+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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+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({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBIPv6ExternalServiceIT {
+
+ private static final String USER = "root";
+ private static final String PASSWORD = "root";
+
+ private static String previousTestNodeAddress;
+
+ @BeforeClass
+ public static void setUp() {
+ IPv6TestUtils.assumeIPv6LoopbackAvailable();
+ previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
+
EnvFactory.getEnv().getConfig().getDataNodeConfig().setEnableRestService(true);
+ EnvFactory.getEnv()
+ .getConfig()
+ .getDataNodeConfig()
+ .setEnableMQTTService(true)
+ .setMqttPayloadFormatter("json");
+ EnvFactory.getEnv().initClusterEnvironment(1, 1);
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ try {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ } finally {
+ IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
+ }
+ }
+
+ @Test
+ public void restServiceCanCommunicateThroughIPv6Loopback() throws Exception {
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
+ final String restAddress =
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(),
dataNode.getRestServicePort());
+
+ try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .untilAsserted(() -> assertRestPingSucceeds(httpClient,
restAddress));
+
+ executeRestNonQuery(
+ httpClient, restAddress, "CREATE TIMESERIES root.ipv6_rest.d1.s1
WITH DATATYPE=INT64");
+ executeRestNonQuery(
+ httpClient, restAddress, "INSERT INTO root.ipv6_rest.d1(time, s1)
VALUES (1, 100)");
+ }
+
+ try (ISession session = EnvFactory.getEnv().getSessionConnection();
+ SessionDataSet dataSet =
+ session.executeQueryStatement("SELECT s1 FROM root.ipv6_rest.d1"))
{
+ assertTrue(dataSet.hasNext());
+ final RowRecord row = dataSet.next();
+ assertEquals(1L, row.getTimestamp());
+ assertEquals(100L, row.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+ }
+
+ @Test
+ public void mqttServiceCanCommunicateThroughIPv6Loopback() throws Exception {
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
+ final String mqttAddress =
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(),
dataNode.getMqttPort());
+
+ final MQTT mqtt = new MQTT();
+ mqtt.setHost("tcp://" + mqttAddress);
+ mqtt.setUserName(USER);
+ mqtt.setPassword(PASSWORD);
+ mqtt.setConnectAttemptsMax(3);
+ mqtt.setReconnectDelay(1000);
+ mqtt.setClientId("ipv6Client");
+
+ final byte[] payload =
+ ("{"
+ + "\"device\":\"root.ipv6_mqtt.d1\","
+ + "\"timestamp\":1,"
+ + "\"measurements\":[\"s1\"],"
+ + "\"values\":[200.0]"
+ + "}")
+ .getBytes(StandardCharsets.UTF_8);
+ final BlockingConnection connection = mqtt.blockingConnection();
+ try {
+ connection.connect();
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ connection.publish("root.ipv6_mqtt.d1", payload,
QoS.AT_LEAST_ONCE, false);
+ return mqttValueIsVisible(session);
+ });
+ }
+ } finally {
+ if (connection.isConnected()) {
+ connection.disconnect();
+ }
+ }
+ }
+
+ private static void assertRestPingSucceeds(
+ final CloseableHttpClient httpClient, final String restAddress) throws
IOException {
+ try (CloseableHttpResponse response =
+ httpClient.execute(new HttpGet("http://"; + restAddress + "/ping"))) {
+ assertEquals(200, response.getStatusLine().getStatusCode());
+ final JsonObject result =
+ JsonParser.parseString(EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8))
+ .getAsJsonObject();
+ assertEquals(200, result.get("code").getAsInt());
+ }
+ }
+
+ private static void executeRestNonQuery(
+ final CloseableHttpClient httpClient, final String restAddress, final
String sql)
+ throws IOException {
+ final HttpPost httpPost = new HttpPost("http://"; + restAddress +
"/rest/v2/nonQuery");
+ httpPost.addHeader("Content-type", "application/json; charset=utf-8");
+ httpPost.setHeader("Accept", "application/json");
+ httpPost.setHeader(
+ "Authorization",
+ Base64.getEncoder()
+ .encodeToString((USER + ":" +
PASSWORD).getBytes(StandardCharsets.UTF_8)));
+ final JsonObject request = new JsonObject();
+ request.addProperty("sql", sql);
+ httpPost.setEntity(new StringEntity(request.toString(),
StandardCharsets.UTF_8));
+
+ try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
+ assertEquals(200, response.getStatusLine().getStatusCode());
+ final JsonObject result =
+ JsonParser.parseString(EntityUtils.toString(response.getEntity(),
StandardCharsets.UTF_8))
+ .getAsJsonObject();
+ assertEquals(200, result.get("code").getAsInt());
+ }
+ }
+
+ private static boolean mqttValueIsVisible(final ISession session) throws
Exception {
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("SELECT s1 FROM root.ipv6_mqtt.d1 WHERE
time = 1")) {
+ if (!dataSet.hasNext()) {
+ return false;
+ }
+ final List<Field> fields = dataSet.next().getFields();
+ return fields.size() == 1 && Math.abs(fields.get(0).getDoubleV() -
200.0) < 0.001;
+ } catch (StatementExecutionException e) {
+ if (e.getMessage() != null && e.getMessage().contains("does not exist"))
{
+ return false;
+ }
+ throw e;
+ }
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java
index e07e6e4ebc1..094b5aebb94 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.isession.SessionConfig;
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 abstract class IoTDBIoTConsensusV23C3DBasicITBase
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 abstract class IoTDBIoTConsensusV23C3DBasicITBase
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 abstract class IoTDBIoTConsensusV23C3DBasicITBase
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 @@ public abstract class IoTDBIoTConsensusV23C3DBasicITBase
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,
diff --git
a/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java
b/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java
new file mode 100644
index 00000000000..aed309c1cce
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java
@@ -0,0 +1,60 @@
+/*
+ * 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.it.utils;
+
+import org.junit.Assume;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+
+import static
org.apache.iotdb.it.env.cluster.ClusterConstant.TEST_NODE_ADDRESS;
+
+public final class IPv6TestUtils {
+
+ public static final String IPV6_LOOPBACK_ADDRESS = "::1";
+
+ private IPv6TestUtils() {
+ // Utility class
+ }
+
+ public static void assumeIPv6LoopbackAvailable() {
+ try (ServerSocket socket = new ServerSocket()) {
+ socket.bind(new
InetSocketAddress(InetAddress.getByName(IPV6_LOOPBACK_ADDRESS), 0));
+ } catch (IOException e) {
+ Assume.assumeNoException("IPv6 loopback is not available", e);
+ }
+ }
+
+ public static String setTestNodeAddressToIPv6Loopback() {
+ final String previousAddress = System.getProperty(TEST_NODE_ADDRESS);
+ System.setProperty(TEST_NODE_ADDRESS, IPV6_LOOPBACK_ADDRESS);
+ return previousAddress;
+ }
+
+ public static void restoreTestNodeAddress(final String previousAddress) {
+ if (previousAddress == null) {
+ System.clearProperty(TEST_NODE_ADDRESS);
+ } else {
+ System.setProperty(TEST_NODE_ADDRESS, previousAddress);
+ }
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
index c99eb5d941a..a5e505f4fd7 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
@@ -130,7 +130,7 @@ public class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
private void testMachineDowntime(String sink) {
StringBuilder a = new StringBuilder();
for (DataNodeWrapper nodeWrapper : receiverEnv.getDataNodeWrapperList()) {
- a.append(nodeWrapper.getIp()).append(":").append(nodeWrapper.getPort());
+ a.append(nodeWrapper.getIpAndPortString());
a.append(",");
}
a.deleteCharAt(a.length() - 1);
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java
new file mode 100644
index 00000000000..716d188c2b1
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java
@@ -0,0 +1,166 @@
+/*
+ * 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.pipe.it.dual.treemodel.auto.basic;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+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.MultiClusterIT2DualTreeAutoBasic;
+import
org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2DualTreeAutoBasic.class})
+public class IoTDBPipeIPv6IT extends AbstractPipeDualTreeModelAutoIT {
+
+ private static String previousTestNodeAddress;
+
+ @BeforeClass
+ public static void setUpIPv6() {
+ IPv6TestUtils.assumeIPv6LoopbackAvailable();
+ previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
+ }
+
+ @AfterClass
+ public static void tearDownIPv6() {
+ IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
+ }
+
+ @Override
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+ setupConfig();
+ senderEnv.initClusterEnvironment(1, 1);
+ receiverEnv.initClusterEnvironment(1, 1);
+ }
+
+ @Override
+ protected void setupConfig() {
+ super.setupConfig();
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setDataReplicationFactor(1)
+ .setSchemaReplicationFactor(1);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setDataReplicationFactor(1)
+ .setSchemaReplicationFactor(1);
+ }
+
+ @Test
+ public void testSyncAndAsyncSinksThroughIPv6Loopback() throws Exception {
+ assertDataNodesUseIPv6(senderEnv.getDataNodeWrapperList());
+ assertDataNodesUseIPv6(receiverEnv.getDataNodeWrapperList());
+
+ final String receiverNodeUrl =
receiverEnv.getDataNodeWrapper(0).getIpAndPortString();
+ Assert.assertTrue(receiverNodeUrl.startsWith("[::1]:"));
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ createAndStartPipe(
+ client,
+ "ipv6_async_pipe",
+ "root.ipv6_async.**",
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ receiverNodeUrl);
+ createAndStartPipe(
+ client,
+ "ipv6_sync_pipe",
+ "root.ipv6_sync.**",
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ receiverNodeUrl);
+ }
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.ipv6_async.d1(time, s1) values (1, 1)",
+ "insert into root.ipv6_sync.d1(time, s1) values (1, 1)",
+ "flush"),
+ null);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(s1) from root.ipv6_async.d1",
+ "count(root.ipv6_async.d1.s1),",
+ Collections.singleton("1,"));
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(s1) from root.ipv6_sync.d1",
+ "count(root.ipv6_sync.d1.s1),",
+ Collections.singleton("1,"));
+ }
+
+ private static void createAndStartPipe(
+ final SyncConfigNodeIServiceClient client,
+ final String pipeName,
+ final String sourcePath,
+ final String sinkName,
+ final String receiverNodeUrl)
+ throws Exception {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put("source.path", sourcePath);
+
+ final Map<String, String> sinkAttributes = new HashMap<>();
+ sinkAttributes.put("sink", sinkName);
+ sinkAttributes.put("sink.batch.enable", "false");
+ sinkAttributes.put("sink.node-urls", receiverNodeUrl);
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq(pipeName,
sinkAttributes).setExtractorAttributes(sourceAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe(pipeName).getCode());
+ }
+
+ private static void assertDataNodesUseIPv6(final Iterable<DataNodeWrapper>
dataNodes) {
+ for (final DataNodeWrapper dataNode : dataNodes) {
+ Assert.assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
+ Assert.assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:"));
+ }
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
index 15ae9fca4c5..529147a77b0 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
@@ -134,7 +134,7 @@ public class IoTDBPipeClusterIT extends
AbstractPipeDualTreeModelAutoIT {
private void testMachineDowntime(String sink) {
StringBuilder a = new StringBuilder();
for (DataNodeWrapper nodeWrapper : receiverEnv.getDataNodeWrapperList()) {
- a.append(nodeWrapper.getIp()).append(":").append(nodeWrapper.getPort());
+ a.append(nodeWrapper.getIpAndPortString());
a.append(",");
}
a.deleteCharAt(a.length() - 1);
diff --git
a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java
new file mode 100644
index 00000000000..32d0d059ac7
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java
@@ -0,0 +1,142 @@
+/*
+ * 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.session.it;
+
+import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.env.EnvFactory;
+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.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.apache.tsfile.read.common.RowRecord;
+import org.junit.AfterClass;
+import org.junit.Assert;
+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.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Collections;
+
+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({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBIPv6ClientIT {
+
+ private static String previousTestNodeAddress;
+
+ @BeforeClass
+ public static void setUp() {
+ IPv6TestUtils.assumeIPv6LoopbackAvailable();
+ previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
+ EnvFactory.getEnv().initClusterEnvironment(1, 1);
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
+ }
+
+ @Test
+ public void clientsCanConnectThroughIPv6Loopback() throws Exception {
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
+ assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:"));
+
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + dataNode.getIpAndPortString(),
+ System.getProperty("User", "root"),
+ System.getProperty("Password", "root"));
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE root.ipv6_client");
+ statement.execute("CREATE TIMESERIES root.ipv6_client.d1.s1 INT64");
+ statement.execute("INSERT INTO root.ipv6_client.d1(time, s1) VALUES (1,
100)");
+ try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.ipv6_client.d1")) {
+ assertTrue(resultSet.next());
+ assertEquals(1L, resultSet.getLong(1));
+ assertEquals(100L, resultSet.getLong(2));
+ assertFalse(resultSet.next());
+ }
+ }
+
+ try (ISession session =
+ EnvFactory.getEnv()
+
.getSessionConnection(Collections.singletonList(dataNode.getIpAndPortString())))
{
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("SELECT s1 FROM root.ipv6_client.d1"))
{
+ Assert.assertTrue(dataSet.hasNext());
+ final RowRecord record = dataSet.next();
+ assertEquals(1L, record.getTimestamp());
+ assertEquals(100L, record.getFields().get(0).getLongV());
+ Assert.assertFalse(dataSet.hasNext());
+ }
+ }
+
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("SELECT s1 FROM root.ipv6_client.d1"))
{
+ Assert.assertTrue(dataSet.hasNext());
+ final RowRecord record = dataSet.next();
+ assertEquals(1L, record.getTimestamp());
+ assertEquals(100L, record.getFields().get(0).getLongV());
+ Assert.assertFalse(dataSet.hasNext());
+ }
+ }
+
+ verifyTableSession(dataNode);
+ }
+
+ private static void verifyTableSession(final DataNodeWrapper dataNode)
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (ITableSession session =
+ EnvFactory.getEnv()
+
.getTableSessionConnection(Collections.singletonList(dataNode.getIpAndPortString())))
{
+ session.executeNonQueryStatement("CREATE DATABASE IF NOT EXISTS
ipv6_table");
+ session.executeNonQueryStatement("USE ipv6_table");
+ session.executeNonQueryStatement("CREATE TABLE table1(tag1 STRING TAG,
s1 INT64 FIELD)");
+ session.executeNonQueryStatement("INSERT INTO table1(time, tag1, s1)
VALUES (1, 'd1', 200)");
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("SELECT time, s1 FROM table1 WHERE
tag1 = 'd1'")) {
+ Assert.assertTrue(dataSet.hasNext());
+ final RowRecord record = dataSet.next();
+ assertEquals(1L, record.getFields().get(0).getLongV());
+ assertEquals(200L, record.getFields().get(1).getLongV());
+ Assert.assertFalse(dataSet.hasNext());
+ }
+ }
+ }
+}
diff --git a/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp
b/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp
index 604099a82d1..55fd13a08f4 100644
--- a/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp
+++ b/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp
@@ -17,6 +17,7 @@
* under the License.
*/
#include "NodesSupplier.h"
+#include "RpcCommon.h"
#include "Session.h"
#include "SessionDataSet.h"
#include <algorithm>
@@ -188,7 +189,7 @@ std::vector<TEndPoint>
NodesSupplier::fetchLatestEndpoints() {
port = record->fields.at(columnPortIdx).intV.value();
}
- if (ip == "0.0.0.0") {
+ if (UrlUtils::isWildcardAddress(ip)) {
log_warn("Skipping invalid node: " + ip + ":" +
std::to_string(port));
continue;
}
@@ -230,4 +231,4 @@ void NodesSupplier::stopBackgroundRefresh() noexcept {
refreshThread_.join();
}
}
-}
\ No newline at end of file
+}
diff --git a/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp
b/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp
index c83b39a5fb5..18c97b4420c 100644
--- a/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp
+++ b/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp
@@ -212,3 +212,26 @@ TEndPoint UrlUtils::parseTEndPointIpv4AndIpv6Url(const
string& endPointUrl) {
return endPoint;
}
+
+bool UrlUtils::isWildcardAddress(const string& host) {
+ if (host.empty()) {
+ return false;
+ }
+
+ const bool bracketed = host.size() >= 2 && host.front() == '[' &&
host.back() == ']';
+ const size_t begin = bracketed ? 1 : 0;
+ const size_t end = bracketed ? host.size() - 1 : host.size();
+ if (end - begin == 7 && host.compare(begin, end - begin, "0.0.0.0") == 0) {
+ return true;
+ }
+
+ bool hasColon = false;
+ for (size_t index = begin; index < end; ++index) {
+ if (host[index] == ':') {
+ hasColon = true;
+ } else if (host[index] != '0') {
+ return false;
+ }
+ }
+ return hasColon;
+}
diff --git a/iotdb-client/client-cpp/src/rpc/RpcCommon.h
b/iotdb-client/client-cpp/src/rpc/RpcCommon.h
index 721150bd6ca..48bff46ffe6 100644
--- a/iotdb-client/client-cpp/src/rpc/RpcCommon.h
+++ b/iotdb-client/client-cpp/src/rpc/RpcCommon.h
@@ -76,6 +76,8 @@ private:
public:
static TEndPoint parseTEndPointIpv4AndIpv6Url(const std::string&
endPointUrl);
+
+ static bool isWildcardAddress(const std::string& host);
};
#endif
diff --git a/iotdb-client/client-cpp/src/session/Session.cpp
b/iotdb-client/client-cpp/src/session/Session.cpp
index 7dd0a7813ed..cf429af0c4d 100644
--- a/iotdb-client/client-cpp/src/session/Session.cpp
+++ b/iotdb-client/client-cpp/src/session/Session.cpp
@@ -2096,7 +2096,7 @@ void Session::Impl::handleQueryRedirection(TEndPoint
endPoint) {
void Session::Impl::handleRedirection(const std::string& deviceId, TEndPoint
endPoint) {
if (!enableRedirection_)
return;
- if (endPoint.ip == "0.0.0.0")
+ if (UrlUtils::isWildcardAddress(endPoint.ip))
return;
getDefaultSessionConnection();
deviceIdToEndpoint[deviceId] = endPoint;
@@ -2122,7 +2122,7 @@ void Session::Impl::handleRedirection(const
std::shared_ptr<storage::IDeviceID>&
TEndPoint endPoint) {
if (!enableRedirection_)
return;
- if (endPoint.ip == "0.0.0.0")
+ if (UrlUtils::isWildcardAddress(endPoint.ip))
return;
getDefaultSessionConnection();
tableModelDeviceIdToEndpoint[deviceId] = endPoint;
diff --git a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp
b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp
index 97c078d9d59..45624f0300d 100644
--- a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp
+++ b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp
@@ -907,6 +907,21 @@ TEST_CASE("UrlUtils - parseTEndPointIpv4AndIpv6Url",
"[UrlUtils]") {
}
}
+TEST_CASE("UrlUtils - isWildcardAddress", "[UrlUtils]") {
+ REQUIRE(UrlUtils::isWildcardAddress("0.0.0.0"));
+ REQUIRE(UrlUtils::isWildcardAddress("[0.0.0.0]"));
+ REQUIRE(UrlUtils::isWildcardAddress("::"));
+ REQUIRE(UrlUtils::isWildcardAddress("[::]"));
+ REQUIRE(UrlUtils::isWildcardAddress("0:0:0:0:0:0:0:0"));
+ REQUIRE(UrlUtils::isWildcardAddress("0::0"));
+
+ REQUIRE_FALSE(UrlUtils::isWildcardAddress(""));
+ REQUIRE_FALSE(UrlUtils::isWildcardAddress("127.0.0.1"));
+ REQUIRE_FALSE(UrlUtils::isWildcardAddress("localhost"));
+ REQUIRE_FALSE(UrlUtils::isWildcardAddress("::1"));
+ REQUIRE_FALSE(UrlUtils::isWildcardAddress("[::1]"));
+}
+
TEST_CASE("TsBlock deserialize rejects truncated malicious payload",
"[TsBlockDeserialize]") {
std::string data(18, '\0');
data[3] = '\x10';
diff --git a/iotdb-client/client-cpp/test/main.cpp
b/iotdb-client/client-cpp/test/main.cpp
index 1bc3425882f..474cb552abb 100644
--- a/iotdb-client/client-cpp/test/main.cpp
+++ b/iotdb-client/client-cpp/test/main.cpp
@@ -30,6 +30,9 @@ struct SessionListener : Catch::TestEventListenerBase {
using TestEventListenerBase::TestEventListenerBase;
void testCaseStarting(Catch::TestCaseInfo const& testInfo) override {
+ if (testInfo.tagsAsString().find("[UrlUtils]") != std::string::npos) {
+ return;
+ }
if (!session) {
SessionBuilder builder;
session = builder.host("127.0.0.1")
@@ -55,4 +58,4 @@ struct SessionListener : Catch::TestEventListenerBase {
}
};
-CATCH_REGISTER_LISTENER(SessionListener)
\ No newline at end of file
+CATCH_REGISTER_LISTENER(SessionListener)
diff --git a/iotdb-client/client-py/iotdb/Session.py
b/iotdb-client/client-py/iotdb/Session.py
index 0a5cb43fd54..e4ed38ab228 100644
--- a/iotdb-client/client-py/iotdb/Session.py
+++ b/iotdb-client/client-py/iotdb/Session.py
@@ -16,48 +16,50 @@
# under the License.
#
+import ipaddress
import logging
import random
-import sys
import struct
+import sys
import warnings
+
+from iotdb.utils.SessionDataSet import SessionDataSet
from thrift.protocol import TBinaryProtocol, TCompactProtocol
from thrift.transport import TSocket, TTransport
from tzlocal import get_localzone_name
-from iotdb.utils.SessionDataSet import SessionDataSet
from .template.Template import Template
from .template.TemplateQueryType import TemplateQueryType
from .thrift.common.ttypes import TEndPoint
from .thrift.rpc.IClientRPCService import (
Client,
- TSCreateTimeseriesReq,
+ TSAppendSchemaTemplateReq,
+ TSCloseSessionReq,
TSCreateAlignedTimeseriesReq,
+ TSCreateMultiTimeseriesReq,
+ TSCreateSchemaTemplateReq,
+ TSCreateTimeseriesReq,
+ TSDropSchemaTemplateReq,
+ TSExecuteStatementReq,
TSInsertRecordReq,
+ TSInsertRecordsOfOneDeviceReq,
+ TSInsertRecordsReq,
TSInsertStringRecordReq,
TSInsertTabletReq,
- TSExecuteStatementReq,
- TSOpenSessionReq,
- TSCreateMultiTimeseriesReq,
- TSCloseSessionReq,
TSInsertTabletsReq,
- TSInsertRecordsReq,
- TSInsertRecordsOfOneDeviceReq,
- TSCreateSchemaTemplateReq,
- TSDropSchemaTemplateReq,
- TSAppendSchemaTemplateReq,
+ TSOpenSessionReq,
TSPruneSchemaTemplateReq,
+ TSQueryTemplateReq,
TSSetSchemaTemplateReq,
TSUnsetSchemaTemplateReq,
- TSQueryTemplateReq,
)
from .thrift.rpc.ttypes import (
TSDeleteDataReq,
+ TSInsertStringRecordsOfOneDeviceReq,
+ TSLastDataQueryReq,
TSProtocolVersion,
- TSSetTimeZoneReq,
TSRawDataQueryReq,
- TSLastDataQueryReq,
- TSInsertStringRecordsOfOneDeviceReq,
+ TSSetTimeZoneReq,
)
from .tsfile.utils.date_utils import parse_date_to_int
from .utils import rpc_utils
@@ -67,6 +69,48 @@ logger = logging.getLogger("IoTDB")
warnings.simplefilter("always", DeprecationWarning)
+def _parse_endpoint_url(endpoint_url):
+ if endpoint_url.startswith("["):
+ end_index = endpoint_url.find("]")
+ if end_index <= 1 or end_index + 1 >= len(endpoint_url):
+ raise RuntimeError("invalid node url: {}".format(endpoint_url))
+ if endpoint_url[end_index + 1] != ":":
+ raise RuntimeError("invalid node url: {}".format(endpoint_url))
+ return endpoint_url[1:end_index], _parse_endpoint_port(
+ endpoint_url, endpoint_url[end_index + 2 :]
+ )
+
+ if "[" in endpoint_url or "]" in endpoint_url:
+ raise RuntimeError("invalid node url: {}".format(endpoint_url))
+ split = endpoint_url.rsplit(":", 1)
+ if len(split) != 2:
+ raise RuntimeError("invalid node url: {}".format(endpoint_url))
+ return split[0], _parse_endpoint_port(endpoint_url, split[1])
+
+
+def _parse_endpoint_port(endpoint_url, port):
+ try:
+ return int(port)
+ except ValueError as e:
+ raise RuntimeError("invalid node url: {}".format(endpoint_url)) from e
+
+
+def _is_wildcard_address(host):
+ if host is None:
+ return False
+ normalized_host = host
+ if normalized_host.startswith("[") and normalized_host.endswith("]"):
+ normalized_host = normalized_host[1:-1]
+ if normalized_host == "0.0.0.0":
+ return True
+ if ":" not in normalized_host:
+ return False
+ try:
+ return ipaddress.ip_address(normalized_host).is_unspecified
+ except ValueError:
+ return False
+
+
class Session(object):
DEFAULT_FETCH_SIZE = 5000
DEFAULT_USER = "root"
@@ -158,9 +202,9 @@ class Session(object):
session.__hosts = []
session.__ports = []
for node_url in node_urls:
- split = node_url.split(":")
- session.__hosts.append(split[0])
- session.__ports.append(int(split[1]))
+ host, port = _parse_endpoint_url(node_url)
+ session.__hosts.append(host)
+ session.__ports.append(port)
session.__host = session.__hosts[0]
session.__port = session.__ports[0]
session.__default_endpoint = TEndPoint(session.__host, session.__port)
@@ -259,6 +303,7 @@ class Session(object):
def __get_transport(self, endpoint):
if self.__use_ssl:
import ssl
+
from thrift.transport import TSSLSocket
if sys.version_info >= (3, 10):
@@ -1986,7 +2031,7 @@ class Session(object):
def handle_redirection(self, device_id, endpoint: TEndPoint):
if self.__enable_redirection:
- if endpoint.ip == "0.0.0.0":
+ if _is_wildcard_address(endpoint.ip):
return 0
if (
device_id not in self.__device_id_to_endpoint
diff --git a/iotdb-client/client-py/tests/unit/test_session_node_url.py
b/iotdb-client/client-py/tests/unit/test_session_node_url.py
new file mode 100644
index 00000000000..d7a59cb4275
--- /dev/null
+++ b/iotdb-client/client-py/tests/unit/test_session_node_url.py
@@ -0,0 +1,85 @@
+# 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.
+#
+
+import pytest
+from iotdb.Session import Session, _is_wildcard_address
+
+
+def test_init_from_node_urls_with_bracketed_ipv6():
+ session = Session.init_from_node_urls(["[::1]:6667"])
+
+ assert session._Session__hosts == ["::1"]
+ assert session._Session__ports == [6667]
+
+
+def test_init_from_node_urls_with_legacy_ipv6():
+ session = Session.init_from_node_urls(["::1:6667"])
+
+ assert session._Session__hosts == ["::1"]
+ assert session._Session__ports == [6667]
+
+
+def test_init_from_node_urls_with_hostname():
+ session = Session.init_from_node_urls(["localhost:6667"])
+
+ assert session._Session__hosts == ["localhost"]
+ assert session._Session__ports == [6667]
+
+
[email protected](
+ "node_url",
+ [
+ "[::1]6667",
+ "[::1]:",
+ "[::1:6667",
+ ],
+)
+def test_init_from_node_urls_rejects_malformed_bracketed_ipv6(node_url):
+ with pytest.raises(RuntimeError):
+ Session.init_from_node_urls([node_url])
+
+
[email protected](
+ "host",
+ [
+ "0.0.0.0",
+ "[0.0.0.0]",
+ "::",
+ "[::]",
+ "0:0:0:0:0:0:0:0",
+ "0::0",
+ ],
+)
+def test_wildcard_address(host):
+ assert _is_wildcard_address(host)
+
+
[email protected](
+ "host",
+ [
+ None,
+ "127.0.0.1",
+ "192.0.2.1",
+ "localhost",
+ "example.com",
+ "::1",
+ "[::1]",
+ ],
+)
+def test_not_wildcard_address(host):
+ assert not _is_wildcard_address(host)
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
index 412093f7386..49aa0860a39 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
@@ -60,41 +60,58 @@ public class Utils {
String suffixURL = null;
if (url.startsWith(Config.IOTDB_URL_PREFIX)) {
String subURL = url.substring(Config.IOTDB_URL_PREFIX.length());
- int i = subURL.lastIndexOf(COLON);
- host = subURL.substring(0, i);
- params.setHost(host);
- i++;
- // parse port
- int port = 0;
- for (; i < subURL.length() && Character.isDigit(subURL.charAt(i)); i++) {
- port = port * 10 + (subURL.charAt(i) - '0');
+ int i = -1;
+ if (subURL.startsWith("[")) {
+ int endIndex = subURL.indexOf("]");
+ if (endIndex > 1
+ && endIndex + 1 < subURL.length()
+ && COLON.equals(subURL.substring(endIndex + 1, endIndex + 2))) {
+ host = subURL.substring(1, endIndex);
+ i = endIndex + 2;
+ }
+ } else if (!subURL.contains("[") && !subURL.contains("]")) {
+ i = subURL.lastIndexOf(COLON);
+ if (i > 0) {
+ host = subURL.substring(0, i);
+ i++;
+ }
}
- suffixURL = i < subURL.length() ? subURL.substring(i) : "";
- // legal port
- if (port >= 1 && port <= 65535) {
- params.setPort(port);
-
- // parse database
- if (i < subURL.length() && subURL.charAt(i) == SLASH) {
- int endIndex = subURL.indexOf(PARAMETER_SEPARATOR, i + 1);
- String database;
- if (endIndex <= i + 1) {
- if (i + 1 == subURL.length()) {
- database = null;
+
+ if (i > 0) {
+ params.setHost(host);
+ // parse port
+ int port = 0;
+ int portStartIndex = i;
+ for (; i < subURL.length() && Character.isDigit(subURL.charAt(i));
i++) {
+ port = port * 10 + (subURL.charAt(i) - '0');
+ }
+ suffixURL = i < subURL.length() ? subURL.substring(i) : "";
+ // legal port
+ if (i > portStartIndex && port >= 1 && port <= 65535) {
+ params.setPort(port);
+
+ // parse database
+ if (i < subURL.length() && subURL.charAt(i) == SLASH) {
+ int endIndex = subURL.indexOf(PARAMETER_SEPARATOR, i + 1);
+ String database;
+ if (endIndex <= i + 1) {
+ if (i + 1 == subURL.length()) {
+ database = null;
+ } else {
+ database = subURL.substring(i + 1);
+ }
+ suffixURL = "";
} else {
- database = subURL.substring(i + 1);
+ database = subURL.substring(i + 1, endIndex);
+ suffixURL = subURL.substring(endIndex);
}
- suffixURL = "";
- } else {
- database = subURL.substring(i + 1, endIndex);
- suffixURL = subURL.substring(endIndex);
+ params.setDb(database);
}
- params.setDb(database);
- }
- matcher = SUFFIX_URL_PATTERN.matcher(suffixURL);
- if (matcher.matches() && parseUrlParam(subURL, info)) {
- isUrlLegal = true;
+ matcher = SUFFIX_URL_PATTERN.matcher(suffixURL);
+ if (matcher.matches() && parseUrlParam(subURL, info)) {
+ isUrlLegal = true;
+ }
}
}
}
diff --git
a/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
b/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
index a84010ad4ec..cd350a539ce 100644
--- a/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
+++ b/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
@@ -27,6 +27,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.util.Optional;
import java.util.Properties;
import static org.junit.Assert.assertEquals;
@@ -90,6 +91,38 @@ public class UtilsTest {
assertEquals(params.getPassword(), userPwd);
}
+ @Test
+ public void testParseBracketedIPV6URL() throws IoTDBURLException {
+ String host = "AD80:E32B:CA25:B3AE:DA4A:DAAF:EEAE:BBBE";
+ int port = 6667;
+ Properties properties = new Properties();
+
+ IoTDBConnectionParams params =
+ Utils.parseUrl(String.format(Config.IOTDB_URL_PREFIX + "[%s]:%s/",
host, port), properties);
+ assertEquals(host, params.getHost());
+ assertEquals(port, params.getPort());
+
+ params =
+ Utils.parseUrl(String.format(Config.IOTDB_URL_PREFIX + "[%s]:%s",
host, port), properties);
+ assertEquals(host, params.getHost());
+ assertEquals(port, params.getPort());
+ }
+
+ @Test(expected = IoTDBURLException.class)
+ public void testParseBracketedIPV6URLWithoutPortSeparator() throws
IoTDBURLException {
+ Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1]6667", new Properties());
+ }
+
+ @Test(expected = IoTDBURLException.class)
+ public void testParseBracketedIPV6URLWithoutPort() throws IoTDBURLException {
+ Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1]:", new Properties());
+ }
+
+ @Test(expected = IoTDBURLException.class)
+ public void testParseBracketedIPV6URLWithoutEndMark() throws
IoTDBURLException {
+ Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1:6667", new Properties());
+ }
+
@Test(expected = IoTDBURLException.class)
public void testParseWrongUrl1() throws IoTDBURLException {
Properties properties = new Properties();
@@ -198,4 +231,17 @@ public class UtilsTest {
assertEquals("/tmp/url-keystore.p12", params.getKeyStore());
assertEquals("url_key_pass", params.getKeyStorePwd());
}
+
+ @Test
+ public void testParseSslConfigFromBracketedIpv6Url() throws
IoTDBURLException {
+ IoTDBConnectionParams params =
+ Utils.parseUrl(
+ "jdbc:iotdb://[::1]:6667/ipv6_db?use_ssl=true&sql_dialect=table",
new Properties());
+
+ assertEquals("::1", params.getHost());
+ assertEquals(6667, params.getPort());
+ assertEquals(Optional.of("ipv6_db"), params.getDb());
+ assertEquals(Constant.TABLE_DIALECT, params.getSqlDialect());
+ assertTrue(params.isUseSSL());
+ }
}
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java
index 39b2390a736..2a02a365d2a 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java
@@ -21,10 +21,14 @@ package org.apache.iotdb.rpc;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
/** The UrlUtils */
public class UrlUtils {
private static final String PORT_SEPARATOR = ":";
- private static final String ABB_COLON = "[";
+ private static final String IPV6_BEGIN_MARK = "[";
+ private static final String IPV6_END_MARK = "]";
private UrlUtils() {}
@@ -32,22 +36,79 @@ public class UrlUtils {
* Parse TEndPoint from a given TEndPointUrl
* example:[D80:0000:0000:0000:ABAA:0000:00C2:0002]:22227
*
- * @param endPointUrl ip:port
- * @return TEndPoint null if parse error
+ * @param endPointUrl host:port or [ipv6]:port
+ * @return parsed TEndPoint
+ * @throws NumberFormatException if the bracketed endpoint format is invalid
or the port is not a
+ * number
*/
public static TEndPoint parseTEndPointIpv4AndIpv6Url(String endPointUrl) {
TEndPoint endPoint = new TEndPoint();
- if (endPointUrl.contains(PORT_SEPARATOR)) {
- int point_position = endPointUrl.lastIndexOf(PORT_SEPARATOR);
- String port =
endPointUrl.substring(endPointUrl.lastIndexOf(PORT_SEPARATOR) + 1);
- String ip = endPointUrl.substring(0, point_position);
- // If the ip/host part is provided as IPv6 address, cut off the
surrounding square brackets.
- if (ip.contains(ABB_COLON)) {
- ip = ip.substring(1, ip.length() - 1);
+ if (endPointUrl.startsWith(IPV6_BEGIN_MARK)) {
+ int endIndex = endPointUrl.indexOf(IPV6_END_MARK);
+ if (endIndex <= 1
+ || endIndex + 1 >= endPointUrl.length()
+ || !PORT_SEPARATOR.equals(endPointUrl.substring(endIndex + 1,
endIndex + 2))) {
+ throw new NumberFormatException();
}
- endPoint.setIp(ip);
- endPoint.setPort(Integer.parseInt(port));
+ endPoint.setIp(endPointUrl.substring(1, endIndex));
+ endPoint.setPort(Integer.parseInt(endPointUrl.substring(endIndex + 2)));
+ return endPoint;
+ }
+ if (endPointUrl.contains(IPV6_BEGIN_MARK) ||
endPointUrl.contains(IPV6_END_MARK)) {
+ throw new NumberFormatException();
+ }
+ int separatorIndex = endPointUrl.lastIndexOf(PORT_SEPARATOR);
+ if (separatorIndex <= 0 || separatorIndex == endPointUrl.length() - 1) {
+ throw new NumberFormatException();
}
+ endPoint.setIp(endPointUrl.substring(0, separatorIndex));
+ endPoint.setPort(Integer.parseInt(endPointUrl.substring(separatorIndex +
1)));
return endPoint;
}
+
+ /**
+ * Convert TEndPoint to a URL string. IPv6 literals are surrounded by square
brackets so the last
+ * colon remains an unambiguous port separator.
+ */
+ public static String convertTEndPointIpv4AndIpv6Url(TEndPoint endPoint) {
+ return formatTEndPointIpv4AndIpv6Url(endPoint.getIp(), endPoint.getPort());
+ }
+
+ /** Format host and port as host:port or [ipv6]:port. This method expects
host without a port. */
+ public static String formatTEndPointIpv4AndIpv6Url(String host, int port) {
+ String formattedHost = host;
+ if (isIpv6Literal(host) && !isBracketedIpv6Literal(host)) {
+ formattedHost = IPV6_BEGIN_MARK + host + IPV6_END_MARK;
+ }
+ return formattedHost + PORT_SEPARATOR + port;
+ }
+
+ public static boolean isWildcardAddress(String host) {
+ if (host == null) {
+ return false;
+ }
+ String normalizedHost = host;
+ if (isBracketedIpv6Literal(host)) {
+ normalizedHost = host.substring(1, host.length() - 1);
+ }
+ if ("0.0.0.0".equals(normalizedHost)) {
+ return true;
+ }
+ if (!isIpv6Literal(normalizedHost)) {
+ return false;
+ }
+ try {
+ return InetAddress.getByName(normalizedHost).isAnyLocalAddress();
+ } catch (UnknownHostException e) {
+ return false;
+ }
+ }
+
+ private static boolean isIpv6Literal(String host) {
+ return host.contains(PORT_SEPARATOR);
+ }
+
+ private static boolean isBracketedIpv6Literal(String host) {
+ return host.startsWith(IPV6_BEGIN_MARK) && host.endsWith(IPV6_END_MARK);
+ }
}
diff --git
a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java
index 1d206973d13..fb1b5b1359b 100644
---
a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java
+++
b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
public class UrlUtilsTest {
@@ -51,6 +53,36 @@ public class UrlUtilsTest {
assertEquals(endPoint.getPort(), 22227);
}
+ @Test(expected = NumberFormatException.class)
+ public void testParseBracketedIPV6URLWithoutPortSeparator() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0]22227");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testParseBracketedIPV6URLWithoutPort() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0]:");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testParseBracketedIPV6URLWithoutEndMark() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0:22227");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testParseURLWithoutPortSeparator() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url("localhost");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testParseURLWithoutHost() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url(":22227");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testParseURLWithoutPort() {
+ UrlUtils.parseTEndPointIpv4AndIpv6Url("localhost:");
+ }
+
@Test
public void testParseIPV4URL() {
String hostAndPoint = "192.0.0.1:22227";
@@ -58,4 +90,52 @@ public class UrlUtilsTest {
assertEquals(endPoint.getIp(), "192.0.0.1");
assertEquals(endPoint.getPort(), 22227);
}
+
+ @Test
+ public void testConvertIPV4URL() {
+ assertEquals(
+ "192.0.0.1:22227",
+ UrlUtils.convertTEndPointIpv4AndIpv6Url(new TEndPoint("192.0.0.1",
22227)));
+ }
+
+ @Test
+ public void testConvertHostNameURL() {
+ assertEquals("localhost:22227",
UrlUtils.formatTEndPointIpv4AndIpv6Url("localhost", 22227));
+ }
+
+ @Test
+ public void testConvertIPV6URL() {
+ assertEquals(
+ "[D80:0000:0000:0000:ABAA:00BB:EEAA:BBDD]:22227",
+ UrlUtils.convertTEndPointIpv4AndIpv6Url(
+ new TEndPoint("D80:0000:0000:0000:ABAA:00BB:EEAA:BBDD", 22227)));
+ }
+
+ @Test
+ public void testConvertIPV6AbbURL() {
+ assertEquals(
+ "[D80::ABAA:0]:22227",
+ UrlUtils.convertTEndPointIpv4AndIpv6Url(new TEndPoint("D80::ABAA:0",
22227)));
+ }
+
+ @Test
+ public void testConvertBracketedIPV6URL() {
+ assertEquals(
+ "[D80::ABAA:0]:22227",
UrlUtils.formatTEndPointIpv4AndIpv6Url("[D80::ABAA:0]", 22227));
+ }
+
+ @Test
+ public void testWildcardAddress() {
+ assertTrue(UrlUtils.isWildcardAddress("0.0.0.0"));
+ assertTrue(UrlUtils.isWildcardAddress("::"));
+ assertTrue(UrlUtils.isWildcardAddress("[::]"));
+ assertTrue(UrlUtils.isWildcardAddress("0:0:0:0:0:0:0:0"));
+ assertTrue(UrlUtils.isWildcardAddress("0::0"));
+
+ assertFalse(UrlUtils.isWildcardAddress(null));
+ assertFalse(UrlUtils.isWildcardAddress("127.0.0.1"));
+ assertFalse(UrlUtils.isWildcardAddress("localhost"));
+ assertFalse(UrlUtils.isWildcardAddress("::1"));
+ assertFalse(UrlUtils.isWildcardAddress("[::1]"));
+ }
}
diff --git
a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java
b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java
index 2ade601350a..69b87b2db16 100644
---
a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java
+++
b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java
@@ -67,6 +67,8 @@ public final class SessionMessages {
// SessionUtils
public static final String NODE_URLS_IS_NULL = "nodeUrls is null";
+ public static final String EXCEPTION_NODEURL_INCORRECT_FORMAT_C1463B2C =
+ "NodeUrl Incorrect format";
// InternalNode (template)
public static final String DUPLICATED_CHILD_IN_TEMPLATE =
diff --git
a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java
b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java
index 3d297e3e7d9..2f28590ac88 100644
---
a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java
+++
b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java
@@ -66,6 +66,8 @@ public final class SessionMessages {
// SessionUtils
public static final String NODE_URLS_IS_NULL = "nodeUrls 为 null";
+ public static final String EXCEPTION_NODEURL_INCORRECT_FORMAT_C1463B2C =
+ "NodeUrl 格式不正确";
// InternalNode (template)
public static final String DUPLICATED_CHILD_IN_TEMPLATE =
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java
index c1f43fa30d4..81d83450f72 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.session;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.isession.INodeSupplier;
import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.session.i18n.SessionMessages;
import org.slf4j.Logger;
@@ -257,8 +258,8 @@ public class NodesSupplier implements INodeSupplier,
Runnable {
List<TEndPoint> res = new ArrayList<>();
while (iterator.next()) {
String ip = iterator.getString(IP_COLUMN_NAME);
- // ignore 0.0.0.0
- if (!"0.0.0.0".equals(ip)) {
+ // ignore wildcard addresses
+ if (!UrlUtils.isWildcardAddress(ip)) {
String port = iterator.getString(PORT_COLUMN_NAME);
if (ip != null && port != null) {
res.add(new TEndPoint(ip, Integer.parseInt(port)));
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
index 6819325a9b2..36581b9690c 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.NoValidValueException;
import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.UrlUtils;
import
org.apache.iotdb.service.rpc.thrift.TCreateTimeseriesUsingSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSAppendSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
@@ -1414,7 +1415,7 @@ public class Session implements ISession {
private void handleRedirection(String deviceId, TEndPoint endpoint) {
if (enableRedirection) {
// no need to redirection
- if (endpoint.ip.equals("0.0.0.0")) {
+ if (UrlUtils.isWildcardAddress(endpoint.ip)) {
return;
}
if (!deviceIdToEndpoint.containsKey(deviceId)
@@ -1440,7 +1441,7 @@ public class Session implements ISession {
private void handleRedirection(IDeviceID deviceId, TEndPoint endpoint) {
if (enableRedirection) {
// no need to redirection
- if (endpoint.ip.equals("0.0.0.0")) {
+ if (UrlUtils.isWildcardAddress(endpoint.ip)) {
return;
}
if (!tableModelDeviceIdToEndpoint.containsKey(deviceId)
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index ad6699aaf87..b498784ad2c 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.service.rpc.thrift.IClientRPCService;
import
org.apache.iotdb.service.rpc.thrift.TCreateTimeseriesUsingSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSAggregationQueryReq;
@@ -83,7 +84,6 @@ import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.StringJoiner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
@@ -1314,14 +1314,11 @@ public class SessionConnection {
if (endPointList == null) {
return MSG_RECONNECTION_FAIL;
}
- StringJoiner urls = new StringJoiner(",");
+ List<String> urls = new ArrayList<>();
for (TEndPoint end : endPointList) {
- StringJoiner url = new StringJoiner(":");
- url.add(end.getIp());
- url.add(String.valueOf(end.getPort()));
- urls.add(url.toString());
+ urls.add(UrlUtils.convertTEndPointIpv4AndIpv6Url(end));
}
- return MSG_RECONNECTION_FAIL.concat(urls.toString());
+ return MSG_RECONNECTION_FAIL.concat(String.join(",", urls));
}
@Override
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index 07c0bc4d7e6..b3de497df39 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.isession.template.Template;
import org.apache.iotdb.isession.util.Version;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp;
import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp;
import org.apache.iotdb.session.DummyNodesSupplier;
@@ -426,7 +427,7 @@ public class SessionPool implements ISessionPool {
this.version = version;
this.thriftDefaultBufferSize = thriftDefaultBufferSize;
this.thriftMaxFrameSize = thriftMaxFrameSize;
- this.formattedNodeUrls = String.format("%s:%s", host, port);
+ this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host,
port);
initThreadPool();
initAvailableNodes(Collections.singletonList(new TEndPoint(host, port)));
}
@@ -469,7 +470,7 @@ public class SessionPool implements ISessionPool {
this.version = version;
this.thriftDefaultBufferSize = thriftDefaultBufferSize;
this.thriftMaxFrameSize = thriftMaxFrameSize;
- this.formattedNodeUrls = String.format("%s:%s", host, port);
+ this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host,
port);
this.useSSL = useSSL;
this.trustStore = trustStore;
this.trustStorePwd = trustStorePwd;
@@ -574,7 +575,7 @@ public class SessionPool implements ISessionPool {
this.host = builder.host;
this.port = builder.rpcPort;
this.nodeUrls = null;
- this.formattedNodeUrls = String.format("%s:%s", host, port);
+ this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host,
port);
if (enableAutoFetch) {
initAvailableNodes(Collections.singletonList(new TEndPoint(host,
port)));
} else {
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
index 3fbb0de0f73..26a2e46e59d 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
@@ -487,8 +487,12 @@ public class SessionUtils {
}
List<TEndPoint> endPointsList = new ArrayList<>();
for (String nodeUrl : nodeUrls) {
- TEndPoint endPoint = UrlUtils.parseTEndPointIpv4AndIpv6Url(nodeUrl);
- endPointsList.add(endPoint);
+ try {
+ endPointsList.add(UrlUtils.parseTEndPointIpv4AndIpv6Url(nodeUrl));
+ } catch (NumberFormatException e) {
+ throw new NumberFormatException(
+ SessionMessages.EXCEPTION_NODEURL_INCORRECT_FORMAT_C1463B2C);
+ }
}
return endPointsList;
}
diff --git a/iotdb-core/ainode/iotdb/ainode/core/config.py
b/iotdb-core/ainode/iotdb/ainode/core/config.py
index 4995dda7bf3..53217f6c27b 100644
--- a/iotdb-core/ainode/iotdb/ainode/core/config.py
+++ b/iotdb-core/ainode/iotdb/ainode/core/config.py
@@ -466,13 +466,24 @@ def load_properties(filepath, sep="=", comment_char="#"):
def parse_endpoint_url(endpoint_url: str) -> TEndPoint:
"""Parse TEndPoint from a given endpoint url.
Args:
- endpoint_url: an endpoint url, format: ip:port
+ endpoint_url: an endpoint url, format: host:port or [ipv6]:port
Returns:
TEndPoint
Raises:
- BadNodeUrlError
+ BadNodeUrlException
"""
- split = endpoint_url.split(":")
+ if endpoint_url.startswith("["):
+ end_index = endpoint_url.find("]")
+ if end_index <= 1 or end_index + 1 >= len(endpoint_url):
+ raise BadNodeUrlException(endpoint_url)
+ if endpoint_url[end_index + 1] != ":":
+ raise BadNodeUrlException(endpoint_url)
+ split = [endpoint_url[1:end_index], endpoint_url[end_index + 2 :]]
+ else:
+ if "[" in endpoint_url or "]" in endpoint_url:
+ raise BadNodeUrlException(endpoint_url)
+ split = endpoint_url.rsplit(":", 1)
+
if len(split) != 2:
raise BadNodeUrlException(endpoint_url)
diff --git a/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py
b/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py
index be1dd9bf1c2..bfe4f56d518 100644
--- a/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py
+++ b/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py
@@ -50,6 +50,13 @@ def _cache_enable() -> bool:
return AINodeDescriptor().get_config().get_ain_data_storage_cache_size() > 0
+def _format_endpoint_url(ip: str, port: int) -> str:
+ formatted_ip = ip
+ if ":" in ip and not (ip.startswith("[") and ip.endswith("]")):
+ formatted_ip = f"[{ip}]"
+ return f"{formatted_ip}:{port}"
+
+
class IoTDBTreeModelDataset(BasicDatabaseForecastDataset):
cache = MemoryLRUCache()
@@ -84,7 +91,7 @@ class IoTDBTreeModelDataset(BasicDatabaseForecastDataset):
self.TIME_CONDITION = " where time>=%s and time<%s"
self.session = Session.init_from_node_urls(
- node_urls=[f"{ip}:{port}"],
+ node_urls=[_format_endpoint_url(ip, port)],
user=username,
password=password,
use_ssl=AINodeDescriptor()
@@ -262,7 +269,7 @@ class IoTDBTableModelDataset(BasicDatabaseForecastDataset):
)
table_session_config = TableSessionConfig(
- node_urls=[f"{ip}:{port}"],
+ node_urls=[_format_endpoint_url(ip, port)],
username=username,
password=password,
use_ssl=AINodeDescriptor()
diff --git a/iotdb-core/ainode/resources/conf/iotdb-ainode.properties
b/iotdb-core/ainode/resources/conf/iotdb-ainode.properties
index 5b653d678a6..6a8f02a62f7 100644
--- a/iotdb-core/ainode/resources/conf/iotdb-ainode.properties
+++ b/iotdb-core/ainode/resources/conf/iotdb-ainode.properties
@@ -23,11 +23,12 @@ cluster_name=defaultCluster
# ConfigNode address registered at AINode startup.
# Allow modifications only before starting the service for the first time.
+# Format: address:port or [ipv6]:port, e.g. 127.0.0.1:10710 or [::1]:10710
# Datatype: String
ain_seed_config_node=127.0.0.1:10710
# Used for connection of DataNode/ConfigNode clients
-# Could set 127.0.0.1(for local test) or ipv4 address
+# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname.
# Datatype: String
ain_rpc_address=127.0.0.1
diff --git a/iotdb-core/ainode/tests/test_config.py
b/iotdb-core/ainode/tests/test_config.py
new file mode 100644
index 00000000000..52955c51137
--- /dev/null
+++ b/iotdb-core/ainode/tests/test_config.py
@@ -0,0 +1,71 @@
+# 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.
+#
+import sys
+import types
+import unittest
+from importlib import import_module
+
+
+class TEndPoint:
+ def __init__(self, ip=None, port=None):
+ self.ip = ip
+ self.port = port
+
+
+thrift_module = types.ModuleType("iotdb.thrift")
+common_module = types.ModuleType("iotdb.thrift.common")
+ttypes_module = types.ModuleType("iotdb.thrift.common.ttypes")
+ttypes_module.TEndPoint = TEndPoint
+sys.modules.setdefault("iotdb.thrift", thrift_module)
+sys.modules.setdefault("iotdb.thrift.common", common_module)
+sys.modules.setdefault("iotdb.thrift.common.ttypes", ttypes_module)
+
+config_module = import_module("iotdb.ainode.core.config")
+exception_module = import_module("iotdb.ainode.core.exception")
+parse_endpoint_url = config_module.parse_endpoint_url
+BadNodeUrlException = exception_module.BadNodeUrlException
+
+
+class ParseEndpointUrlTest(unittest.TestCase):
+ def test_parse_bracketed_ipv6_endpoint(self):
+ endpoint = parse_endpoint_url("[::1]:6667")
+
+ self.assertEqual("::1", endpoint.ip)
+ self.assertEqual(6667, endpoint.port)
+
+ def test_parse_legacy_ipv6_endpoint(self):
+ endpoint = parse_endpoint_url("::1:6667")
+
+ self.assertEqual("::1", endpoint.ip)
+ self.assertEqual(6667, endpoint.port)
+
+ def test_reject_malformed_endpoint(self):
+ malformed_urls = [
+ "[]:123",
+ "foo[::1]:6667",
+ "foo]bar:6667",
+ ]
+
+ for endpoint_url in malformed_urls:
+ with self.subTest(endpoint_url=endpoint_url):
+ with self.assertRaises(BadNodeUrlException):
+ parse_endpoint_url(endpoint_url)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
index cb71ed3fa4d..7b052cebf97 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.consensus.config.RatisConfig;
import org.apache.iotdb.consensus.i18n.ConsensusMessages;
import org.apache.iotdb.rpc.AutoScalingBufferWriteTransport;
import org.apache.iotdb.rpc.RpcSslUtils;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.ratis.client.RaftClientConfigKeys;
import org.apache.ratis.conf.Parameters;
@@ -82,7 +83,7 @@ public class Utils {
private Utils() {}
public static String hostAddress(TEndPoint endpoint) {
- return String.format("%s:%d", endpoint.getIp(), endpoint.getPort());
+ return UrlUtils.convertTEndPointIpv4AndIpv6Url(endpoint);
}
public static String fromTEndPointToString(TEndPoint endpoint) {
@@ -103,8 +104,7 @@ public class Utils {
}
public static TEndPoint fromRaftPeerAddressToTEndPoint(String address) {
- String[] items = address.split(":");
- return new TEndPoint(items[0], Integer.parseInt(items[1]));
+ return UrlUtils.parseTEndPointIpv4AndIpv6Url(address);
}
public static int fromRaftPeerIdToNodeId(RaftPeerId id) {
@@ -112,8 +112,7 @@ public class Utils {
}
public static TEndPoint fromRaftPeerProtoToTEndPoint(RaftPeerProto proto) {
- String[] items = proto.getAddress().split(":");
- return new TEndPoint(items[0], Integer.parseInt(items[1]));
+ return fromRaftPeerAddressToTEndPoint(proto.getAddress());
}
// priority is used as ordinal of leader election
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java
index b1ba2195006..5633459d387 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java
@@ -19,11 +19,13 @@
package org.apache.iotdb.consensus.ratis.utils;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.consensus.config.RatisConfig;
import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.util.TimeDuration;
import org.junit.Assert;
import org.junit.Test;
@@ -64,4 +66,56 @@ public class UtilsTest {
TimeDuration.valueOf(52700, TimeUnit.MILLISECONDS),
Utils.getMaxRetrySleepTime(clientConfig));
}
+
+ @Test
+ public void testRaftPeerAddressRoundTripWithIpv4() {
+ TEndPoint endPoint = new TEndPoint("192.0.0.1", 10720);
+
+ Assert.assertEquals("192.0.0.1:10720", Utils.hostAddress(endPoint));
+ Assert.assertEquals(
+ endPoint,
Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint)));
+ }
+
+ @Test
+ public void testRaftPeerAddressRoundTripWithHostName() {
+ TEndPoint endPoint = new TEndPoint("localhost", 10720);
+
+ Assert.assertEquals("localhost:10720", Utils.hostAddress(endPoint));
+ Assert.assertEquals(
+ endPoint,
Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint)));
+ }
+
+ @Test
+ public void testRaftPeerAddressRoundTripWithIpv6() {
+ TEndPoint endPoint = new TEndPoint("::1", 10720);
+
+ Assert.assertEquals("[::1]:10720", Utils.hostAddress(endPoint));
+ Assert.assertEquals(
+ endPoint,
Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint)));
+ Assert.assertEquals(endPoint,
Utils.fromRaftPeerAddressToTEndPoint("::1:10720"));
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testRaftPeerAddressRejectsBracketedIpv6WithoutPortSeparator() {
+ Utils.fromRaftPeerAddressToTEndPoint("[::1]10720");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testRaftPeerAddressRejectsBracketedIpv6WithoutPort() {
+ Utils.fromRaftPeerAddressToTEndPoint("[::1]:");
+ }
+
+ @Test(expected = NumberFormatException.class)
+ public void testRaftPeerAddressRejectsBracketedIpv6WithoutEndMark() {
+ Utils.fromRaftPeerAddressToTEndPoint("[::1:10720");
+ }
+
+ @Test
+ public void testRaftPeerProtoRoundTripWithIpv6() {
+ TEndPoint endPoint = new TEndPoint("0:0:0:0:0:0:0:1", 10720);
+ RaftPeer raftPeer = Utils.fromNodeInfoAndPriorityToRaftPeer(1, endPoint,
0);
+
+ Assert.assertEquals("[0:0:0:0:0:0:0:1]:10720", raftPeer.getAddress());
+ Assert.assertEquals(endPoint,
Utils.fromRaftPeerProtoToTEndPoint(raftPeer.getRaftPeerProto()));
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
index 460c3c5f846..230bc80c98f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
@@ -39,6 +39,7 @@ import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataN
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
import org.apache.thrift.TException;
@@ -404,7 +405,10 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
}
public void updateLeaderCache(final String deviceId, final TEndPoint
endPoint) {
- if (!useLeaderCache || deviceId == null || endPoint == null) {
+ if (!useLeaderCache
+ || deviceId == null
+ || endPoint == null
+ || UrlUtils.isWildcardAddress(endPoint.getIp())) {
return;
}
@@ -561,7 +565,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
"%s-%s",
receiverAttributes,
endPoints.stream()
- .map(endPoint -> String.format("%s:%s", endPoint.getIp(),
endPoint.getPort()))
+ .map(UrlUtils::convertTEndPointIpv4AndIpv6Url)
.distinct()
.sorted()
.collect(Collectors.joining(",", "[", "]")));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
index df966c78acc..e8f081cff00 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
@@ -119,7 +120,10 @@ public class IoTDBDataNodeSyncClientManager extends
IoTDBSyncClientManager
}
public void updateLeaderCache(final String deviceId, final TEndPoint
endPoint) {
- if (!useLeaderCache || deviceId == null || endPoint == null) {
+ if (!useLeaderCache
+ || deviceId == null
+ || endPoint == null
+ || UrlUtils.isWildcardAddress(endPoint.getIp())) {
return;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index a97f4ed3000..9466543d164 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -66,6 +66,7 @@ import
org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import com.google.common.collect.ImmutableSet;
@@ -912,7 +913,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
}
private static String format(final TEndPoint endPoint) {
- return Objects.isNull(endPoint) ? null : endPoint.getIp() + ":" +
endPoint.getPort();
+ return Objects.isNull(endPoint) ? null :
UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint);
}
//////////////////////////// Operations for close
////////////////////////////
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java
index 483bb3ae1ff..e71af67cafd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java
@@ -33,6 +33,7 @@ import
org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
import
org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.UrlUtils;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
@@ -173,7 +174,7 @@ public class TestConnectionTask implements IConfigTask {
}
private static String endPointToString(TEndPoint endPoint) {
- return endPoint.getIp() + ":" + endPoint.getPort();
+ return UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint);
}
private static void sortTestConnectionResp(TTestConnectionResp origin) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
index f564dbbd70d..c26f35ddc77 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java
@@ -22,16 +22,24 @@ package org.apache.iotdb.db.pipe.sink;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager;
+import
org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeCacheLeaderClientManager;
+import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager;
import org.junit.Assert;
import org.junit.Test;
import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.ExecutorService;
public class IoTDBDataNodeAsyncClientManagerTest {
+ private static final List<String> WILDCARD_ADDRESSES =
+ Arrays.asList("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0", "0::0");
+
@Test
public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges()
throws Exception {
final IoTDBDataNodeAsyncClientManager managerWithSkipIf =
@@ -114,6 +122,85 @@ public class IoTDBDataNodeAsyncClientManagerTest {
}
}
+ @Test
+ public void
testAsyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() {
+ final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667);
+ final List<TEndPoint> endPoints = new
ArrayList<>(Collections.singletonList(originalEndPoint));
+ final IoTDBDataNodeAsyncClientManager manager =
createAsyncManager(endPoints);
+
+ try {
+ for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) {
+ final String deviceId = "async-wildcard-device-" + i;
+ manager.updateLeaderCache(deviceId, new
TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i));
+
+ Assert.assertEquals(Collections.singletonList(originalEndPoint),
endPoints);
+ Assert.assertNull(
+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
+ }
+
+ final String ipv6DeviceId = "async-ipv6-device";
+ final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677);
+ manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint);
+
+ Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint),
endPoints);
+ Assert.assertEquals(
+ ipv6EndPoint,
+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
+ ipv6DeviceId));
+ } finally {
+ manager.close();
+ }
+ }
+
+ @Test
+ public void
testSyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() {
+ final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667);
+ final List<TEndPoint> endPoints = new
ArrayList<>(Collections.singletonList(originalEndPoint));
+ final TestIoTDBDataNodeSyncClientManager manager =
+ new TestIoTDBDataNodeSyncClientManager(endPoints);
+
+ try {
+ for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) {
+ final String deviceId = "sync-wildcard-device-" + i;
+ manager.updateLeaderCache(deviceId, new
TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i));
+
+ Assert.assertEquals(Collections.singletonList(originalEndPoint),
endPoints);
+ Assert.assertEquals(0, manager.getReconstructionCount());
+ Assert.assertNull(
+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
+ }
+
+ final String ipv6DeviceId = "sync-ipv6-device";
+ final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677);
+ manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint);
+
+ Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint),
endPoints);
+ Assert.assertEquals(1, manager.getReconstructionCount());
+ Assert.assertEquals(
+ ipv6EndPoint,
+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
+ ipv6DeviceId));
+ } finally {
+ manager.close();
+ }
+ }
+
+ private static IoTDBDataNodeAsyncClientManager createAsyncManager(
+ final List<TEndPoint> endPoints) {
+ return new IoTDBDataNodeAsyncClientManager(
+ endPoints,
+ true,
+ "round-robin",
+ new UserEntity(1L, "user", "cli-host"),
+ "password",
+ true,
+ "sync",
+ true,
+ true,
+ false,
+ true);
+ }
+
private static String getReceiverAttributes(final
IoTDBDataNodeAsyncClientManager manager)
throws Exception {
final Field field =
@@ -142,4 +229,37 @@ public class IoTDBDataNodeAsyncClientManagerTest {
field.setAccessible(true);
return (ExecutorService) field.get(manager);
}
+
+ private static class TestIoTDBDataNodeSyncClientManager extends
IoTDBDataNodeSyncClientManager {
+
+ private int reconstructionCount;
+
+ private TestIoTDBDataNodeSyncClientManager(final List<TEndPoint>
endPoints) {
+ super(
+ endPoints,
+ false,
+ null,
+ null,
+ null,
+ null,
+ true,
+ "round-robin",
+ new UserEntity(1L, "user", "cli-host"),
+ "password",
+ true,
+ "sync",
+ true,
+ true,
+ true);
+ }
+
+ @Override
+ protected void reconstructClient(final TEndPoint endPoint) {
+ reconstructionCount++;
+ }
+
+ private int getReconstructionCount() {
+ return reconstructionCount;
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index c54878bfe1b..bfb98ebadc1 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -36,7 +36,7 @@ cluster_name=defaultCluster
# For other ConfigNodes that to join the cluster, cn_seed_config_node points
to any running ConfigNode's cn_internal_address:cn_internal_port.
# Note: After this ConfigNode successfully joins the cluster for the first
time, this parameter is no longer used.
# Each node automatically maintains the list of ConfigNodes and traverses
connections when restarting.
-# Format: address:port e.g. 127.0.0.1:10710
+# Format: address:port or [ipv6]:port e.g. 127.0.0.1:10710 or [::1]:10710
# effectiveMode: first_start
# Datatype: String
cn_seed_config_node=127.0.0.1:10710
@@ -44,7 +44,7 @@ cn_seed_config_node=127.0.0.1:10710
# dn_seed_config_node points to any running ConfigNode's
cn_internal_address:cn_internal_port.
# Note: After this DataNode successfully joins the cluster for the first time,
this parameter is no longer used.
# Each node automatically maintains the list of ConfigNodes and traverses
connections when restarting.
-# Format: address:port e.g. 127.0.0.1:10710
+# Format: address:port or [ipv6]:port e.g. 127.0.0.1:10710 or [::1]:10710
# effectiveMode: first_start
# Datatype: String
dn_seed_config_node=127.0.0.1:10710
@@ -54,7 +54,7 @@ dn_seed_config_node=127.0.0.1:10710
####################
# Used for RPC communication inside cluster.
-# Could set 127.0.0.1(for local test) or ipv4 address.
+# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname.
# effectiveMode: first_start
# Datatype: String
cn_internal_address=127.0.0.1
@@ -70,7 +70,7 @@ cn_internal_port=10710
cn_consensus_port=10720
# Used for connection of IoTDB native clients(Session)
-# Could set 127.0.0.1(for local test) or ipv4 address
+# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname.
# effectiveMode: restart
# Datatype: String
dn_rpc_address=127.0.0.1
@@ -82,7 +82,7 @@ dn_rpc_address=127.0.0.1
dn_rpc_port=6667
# Used for communication inside cluster.
-# could set 127.0.0.1(for local test) or ipv4 address.
+# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname.
# effectiveMode: first_start
# Datatype: String
dn_internal_address=127.0.0.1
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java
index 0c6aab02566..87d50fc75c7 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java
@@ -49,13 +49,10 @@ public class NodeUrlUtils {
* Convert TEndPoint to TEndPointUrl
*
* @param endPoint TEndPoint
- * @return TEndPointUrl with format ip:port
+ * @return TEndPointUrl with format host:port or [ipv6]:port
*/
public static String convertTEndPointUrl(TEndPoint endPoint) {
- StringJoiner url = new StringJoiner(":");
- url.add(endPoint.getIp());
- url.add(String.valueOf(endPoint.getPort()));
- return url.toString();
+ return UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint);
}
/**
@@ -75,7 +72,7 @@ public class NodeUrlUtils {
/**
* Parse TEndPoint from a given TEndPointUrl
*
- * @param endPointUrl ip:port
+ * @param endPointUrl host:port or [ipv6]:port
* @return TEndPoint
* @throws BadNodeUrlException Throw when unable to parse
*/
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java
index 6f6e86ac8de..df1c6a216d4 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java
@@ -69,6 +69,9 @@ public class NodeUrlUtilsTest {
new TEndPoint("::13.1.68.3", 6669));
final String endPointUrls =
"AD80:E32B:CA25:B3AE:DC4C:DAAF:CCDE:2345:6667,[0:0:0:0:0:FFFF:129.144.52.38]:6668,[::13.1.68.3]:6669";
+ final String convertedEndPointUrls =
+
"[AD80:E32B:CA25:B3AE:DC4C:DAAF:CCDE:2345]:6667,[0:0:0:0:0:FFFF:129.144.52.38]:6668,[::13.1.68.3]:6669";
+ Assert.assertEquals(convertedEndPointUrls,
NodeUrlUtils.convertTEndPointUrls(endPoints));
Assert.assertEquals(endPoints,
NodeUrlUtils.parseTEndPointUrls(endPointUrls));
}
@@ -90,6 +93,10 @@ public class NodeUrlUtilsTest {
new TEndPoint("AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD",
22282)));
final String configNodeUrls =
"0,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22277,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22278;1,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22279,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22280;2,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22281,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22282";
+ final String convertedConfigNodeUrls =
+
"0,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22277,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22278;1,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22279,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22280;2,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22281,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22282";
+ Assert.assertEquals(
+ convertedConfigNodeUrls,
NodeUrlUtils.convertTConfigNodeUrls(configNodeLocations));
Assert.assertEquals(configNodeLocations,
NodeUrlUtils.parseTConfigNodeUrls(configNodeUrls));
}
}
