This is an automated email from the ASF dual-hosted git repository. HTHou pushed a commit to branch codex/ipv6-endpoint-support in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 736c55af84010e10fac8cdcfb8aa1f6baf7ca296 Author: HTHou <[email protected]> AuthorDate: Thu Jul 9 11:41:03 2026 +0800 Support IPv6 endpoint URLs --- .../iotdb/it/env/cluster/env/AbstractEnv.java | 29 ++++---- .../it/env/cluster/node/AbstractNodeWrapper.java | 3 +- .../iotdb/it/env/remote/env/RemoteServerEnv.java | 23 ++++--- .../IoTDBIoTConsensusV23C3DBasicITBase.java | 19 ++++-- .../manual/enhanced/IoTDBPipeClusterIT.java | 2 +- .../auto/enhanced/IoTDBPipeClusterIT.java | 2 +- iotdb-client/client-py/iotdb/Session.py | 66 +++++++++++++------ .../client-py/tests/unit/test_session_node_url.py | 54 +++++++++++++++ .../src/main/java/org/apache/iotdb/jdbc/Utils.java | 77 +++++++++++++--------- .../test/java/org/apache/iotdb/jdbc/UtilsTest.java | 42 ++++++++++++ .../main/java/org/apache/iotdb/rpc/UrlUtils.java | 48 ++++++++++++-- .../java/org/apache/iotdb/rpc/UrlUtilsTest.java | 48 ++++++++++++++ .../apache/iotdb/session/SessionConnection.java | 11 ++-- .../org/apache/iotdb/session/pool/SessionPool.java | 7 +- iotdb-core/ainode/iotdb/ainode/core/config.py | 13 +++- .../ainode/iotdb/ainode/core/ingress/iotdb.py | 11 +++- .../ainode/resources/conf/iotdb-ainode.properties | 3 +- .../apache/iotdb/consensus/ratis/utils/Utils.java | 9 ++- .../iotdb/consensus/ratis/utils/UtilsTest.java | 54 +++++++++++++++ .../client/IoTDBDataNodeAsyncClientManager.java | 3 +- .../thrift/async/IoTDBDataRegionAsyncSink.java | 3 +- .../execution/config/sys/TestConnectionTask.java | 3 +- .../conf/iotdb-system.properties.template | 10 +-- .../apache/iotdb/commons/utils/NodeUrlUtils.java | 9 +-- .../iotdb/commons/utils/NodeUrlUtilsTest.java | 7 ++ 25 files changed, 433 insertions(+), 123 deletions(-) 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 e5d4e19834d..c4534fb5a7e 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; @@ -150,9 +151,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); @@ -162,9 +162,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); @@ -1038,7 +1037,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 @@ -1618,18 +1618,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); @@ -1652,10 +1648,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..409ec744641 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; @@ -606,7 +607,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/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/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/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/iotdb-client/client-py/iotdb/Session.py b/iotdb-client/client-py/iotdb/Session.py index 0a5cb43fd54..a65d40ab2ad 100644 --- a/iotdb-client/client-py/iotdb/Session.py +++ b/iotdb-client/client-py/iotdb/Session.py @@ -18,46 +18,47 @@ 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 +68,32 @@ 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 + + class Session(object): DEFAULT_FETCH_SIZE = 5000 DEFAULT_USER = "root" @@ -158,9 +185,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 +286,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): 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..fbfde69122f --- /dev/null +++ b/iotdb-client/client-py/tests/unit/test_session_node_url.py @@ -0,0 +1,54 @@ +# 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 + + +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]) 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..fa394f91c69 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 @@ -90,6 +90,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 +230,14 @@ 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?use_ssl=true", new Properties()); + + assertEquals("::1", params.getHost()); + assertEquals(6667, params.getPort()); + 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..f0c08e164ce 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 @@ -24,7 +24,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; /** 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 +33,57 @@ public class UrlUtils { * Parse TEndPoint from a given TEndPointUrl * example:[D80:0000:0000:0000:ABAA:0000:00C2:0002]:22227 * - * @param endPointUrl ip:port + * @param endPointUrl host:port or [ipv6]:port * @return TEndPoint null if parse error */ public static TEndPoint parseTEndPointIpv4AndIpv6Url(String endPointUrl) { TEndPoint endPoint = new TEndPoint(); + 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(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(); + } 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); - } endPoint.setIp(ip); endPoint.setPort(Integer.parseInt(port)); } 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; + } + + 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..9c7061321a8 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 @@ -51,6 +51,21 @@ 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 public void testParseIPV4URL() { String hostAndPoint = "192.0.0.1:22227"; @@ -58,4 +73,37 @@ 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)); + } } 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-core/ainode/iotdb/ainode/core/config.py b/iotdb-core/ainode/iotdb/ainode/core/config.py index 4995dda7bf3..fdc56d774cd 100644 --- a/iotdb-core/ainode/iotdb/ainode/core/config.py +++ b/iotdb-core/ainode/iotdb/ainode/core/config.py @@ -466,13 +466,22 @@ 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 """ - 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: + 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/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 381206258f1..247e45d27ec 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; @@ -562,7 +563,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/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 250ebeabb39..685620a60b7 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; @@ -876,7 +877,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/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 05ff507226d..e04e53f5d6d 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)); } }
