diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java index 30cee019f660a..c263f0dba4cfe 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.it.env.cluster.EnvUtils; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic; @@ -39,9 +40,11 @@ import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.io.File; import java.nio.charset.Charset; import java.sql.Connection; import java.sql.ResultSet; +import java.sql.SQLException; import java.sql.Statement; import java.util.Arrays; import java.util.Collections; @@ -474,8 +477,6 @@ private void testLoadTsFileWithoutVerify(final String loadTsFileStrategy) throws try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { - // Do not fail if the failure has nothing to do with pipe - // Because the failures will randomly generate due to resource limitation TestUtils.executeNonQueries( senderEnv, Arrays.asList("insert into root.vehicle.d0(time, s1) values (1, 1)", "flush"), @@ -517,6 +518,115 @@ private void testLoadTsFileWithoutVerify(final String loadTsFileStrategy) throws } } + @Test + public void testSinkPortBinder() throws Exception { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + + final String receiverIp = receiverDataNode.getIp(); + final int receiverPort = receiverDataNode.getPort(); + + try (final Connection connection = senderEnv.getConnection(); + final Statement statement = connection.createStatement()) { + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "22220,22221-2220", + "1107: The port 22221-2220 is invalid, the min value shall be smaller than the max value in the port range"); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "xxx", + "1107: The port xxx is invalid, the port must be an integer"); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "10-20-30", + "1107: The port 10-20-30 is invalid, a port shall either be 'port' or 'minPort-maxPort'."); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "10, ,30", + "1107: The port is invalid, the port must be an integer"); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "100000", + "1107: The port 100000 is invalid, the port must be >= 0 and <= 65535"); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "10-", + "1107: The port 10- is invalid, a port shall either be 'port' or 'minPort-maxPort'."); + assertInvalidSendPorts( + statement, + receiverIp, + receiverPort, + "10,", + "1107: The port is invalid, the port must be an integer"); + + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList("insert into root.vehicle.d0(time, s1) values (1, 1)", "flush"), + null); + + final int[] firstPorts = EnvUtils.searchAvailablePorts(); + final int[] secondPorts = EnvUtils.searchAvailablePorts(); + + try { + statement.execute( + String.format( + "create pipe testPipe ('ip'='%s', 'port'='%s', 'send-ports'='%s,%s-%s,%s-%s')", + receiverIp, + receiverPort, + firstPorts[0], + firstPorts[0], + firstPorts[firstPorts.length - 1], + secondPorts[0], + secondPorts[secondPorts.length - 1])); + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select * from root.vehicle.**", + "Time,root.vehicle.d0.s1,", + Collections.singleton("1,1.0,")); + } finally { + String lockPath = EnvUtils.getLockFilePath(firstPorts[0]); + if (!new File(lockPath).delete()) { + System.out.printf("Delete lock file %s failed%n", lockPath); + } + + lockPath = EnvUtils.getLockFilePath(secondPorts[0]); + if (!new File(lockPath).delete()) { + System.out.printf("Delete lock file %s failed%n", lockPath); + } + } + } + } + + private static void assertInvalidSendPorts( + final Statement statement, + final String receiverIp, + final int receiverPort, + final String sendPorts, + final String expectedMessage) + throws SQLException { + try { + statement.execute( + String.format( + "create pipe testPipe ('ip'='%s', 'port'='%s', 'send-ports'='%s')", + receiverIp, receiverPort, sendPorts)); + Assert.fail("Invalid send-ports value was accepted: " + sendPorts); + } catch (final SQLException e) { + Assert.assertEquals(expectedMessage, e.getMessage()); + } + } + @Test public void testSpecialPartialInsert() throws Exception { try (final Connection connection = senderEnv.getConnection(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java index 25c0bc58b6142..008feae409742 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/client/IoTDBConfigNodeSyncClientManager.java @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV2Req; import org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV1Req; @@ -50,6 +52,7 @@ public IoTDBConfigNodeSyncClientManager( String loadTsFileStrategy, boolean validateTsFile, boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { super( endPoints, @@ -66,6 +69,7 @@ public IoTDBConfigNodeSyncClientManager( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java index 636e28a455e98..0c6ed88172174 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFilePieceReq; @@ -80,6 +82,7 @@ protected IoTDBSyncClientManager constructClient( final String loadTsFileStrategy, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { return new IoTDBConfigNodeSyncClientManager( nodeUrls, @@ -95,6 +98,7 @@ protected IoTDBSyncClientManager constructClient( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); } 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 2b85ac51814e3..020cfb33801b3 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 @@ -29,6 +29,8 @@ import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.client.IoTDBClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant; @@ -106,6 +108,7 @@ public IoTDBDataNodeAsyncClientManager( final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, final boolean isTSFileUsed, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { super( endPoints, @@ -116,13 +119,14 @@ public IoTDBDataNodeAsyncClientManager( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); endPointSet = new HashSet<>(endPoints); receiverAttributes = String.format( - "%s-%s-%s-%s-%s-%s-%s", + "%s-%s-%s-%s-%s-%s-%s-%s", Base64.getEncoder() .encodeToString((userEntity.getUsername() + ":" + password).getBytes()), shouldReceiverConvertOnTypeMismatch, @@ -130,6 +134,7 @@ public IoTDBDataNodeAsyncClientManager( validateTsFile, shouldMarkAsPipeRequest, isTSFileUsed, + candidatePorts, skipIfNoPrivileges); clientResourceKey = generateClientResourceKey(receiverAttributes, endPoints); synchronized (IoTDBDataNodeAsyncClientManager.class) { @@ -139,9 +144,10 @@ public IoTDBDataNodeAsyncClientManager( new IClientManager.Factory() .createClientManager( isTSFileUsed - ? new ClientPoolFactory - .AsyncPipeTsFileDataTransferServiceClientPoolFactory() - : new ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory())); + ? new ClientPoolFactory.AsyncPipeTsFileDataTransferServiceClientPoolFactory( + candidatePorts) + : new ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory( + candidatePorts))); } endPoint2Client = ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get(clientResourceKey); 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 e8f081cff0035..f8022274f3c46 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 @@ -22,6 +22,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV2Req; @@ -62,6 +64,7 @@ public IoTDBDataNodeSyncClientManager( final String loadTsFileStrategy, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { super( endPoints, @@ -78,6 +81,7 @@ public IoTDBDataNodeSyncClientManager( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); } 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 64f2591975026..744b61bc42366 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 @@ -227,6 +227,7 @@ public void customize( loadTsFileValidation, shouldMarkAsPipeRequest, false, + candidatePorts, skipIfNoPrivileges); clientManager.setPipeInfo(pipeName, creationTime); @@ -244,6 +245,7 @@ public void customize( loadTsFileValidation, shouldMarkAsPipeRequest, isSplitTSFileBatchModeEnabled, + candidatePorts, skipIfNoPrivileges); transferTsFileClientManager.setPipeInfo(pipeName, creationTime); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java index ff32bbbb06fb2..b961839bd7b52 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataNodeSyncSink.java @@ -21,6 +21,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSslSyncSink; import org.apache.iotdb.db.conf.IoTDBConfig; @@ -55,6 +57,7 @@ protected IoTDBSyncClientManager constructClient( final String loadTsFileStrategy, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { clientManager = new IoTDBDataNodeSyncClientManager( @@ -72,6 +75,7 @@ protected IoTDBSyncClientManager constructClient( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); return clientManager; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java index 02817fb424528..de1b8b8ec186b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java @@ -40,8 +40,6 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.Collections; import java.util.LinkedHashSet; @@ -54,8 +52,6 @@ public class SchemaValidator { - private static final Logger LOGGER = LoggerFactory.getLogger(SchemaValidator.class); - public static void validate( ISchemaFetcher schemaFetcher, InsertBaseStatement insertStatement, MPPQueryContext context) { try { 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 c26f35ddc77d3..3c9f170cdaa1c 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 @@ -21,6 +21,8 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; 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; @@ -54,6 +56,7 @@ public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges() throws true, true, false, + new IntervalManager<>(), true); final IoTDBDataNodeAsyncClientManager managerWithoutSkipIf = new IoTDBDataNodeAsyncClientManager( @@ -67,6 +70,7 @@ public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges() throws true, true, false, + new IntervalManager<>(), false); try { @@ -94,6 +98,7 @@ public void testClientResourcesShouldDifferentiateEndPoints() throws Exception { true, true, true, + new IntervalManager<>(), true); final IoTDBDataNodeAsyncClientManager secondManager = new IoTDBDataNodeAsyncClientManager( @@ -107,6 +112,7 @@ public void testClientResourcesShouldDifferentiateEndPoints() throws Exception { true, true, true, + new IntervalManager<>(), true); try { @@ -122,6 +128,27 @@ public void testClientResourcesShouldDifferentiateEndPoints() throws Exception { } } + @Test + public void testClientResourcesShouldDifferentiateSendPorts() throws Exception { + final List endPoints = Collections.singletonList(new TEndPoint("127.0.0.1", 6667)); + final IntervalManager firstPorts = new IntervalManager<>(); + firstPorts.addInterval(new PlainInterval(31000, 31000)); + final IntervalManager secondPorts = new IntervalManager<>(); + secondPorts.addInterval(new PlainInterval(31001, 31001)); + final IoTDBDataNodeAsyncClientManager firstManager = createAsyncManager(endPoints, firstPorts); + final IoTDBDataNodeAsyncClientManager secondManager = + createAsyncManager(endPoints, secondPorts); + + try { + Assert.assertNotEquals( + getClientResourceKey(firstManager), getClientResourceKey(secondManager)); + Assert.assertNotSame(getEndPoint2Client(firstManager), getEndPoint2Client(secondManager)); + } finally { + firstManager.close(); + secondManager.close(); + } + } + @Test public void testAsyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() { final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667); @@ -187,6 +214,11 @@ public void testSyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() { private static IoTDBDataNodeAsyncClientManager createAsyncManager( final List endPoints) { + return createAsyncManager(endPoints, new IntervalManager<>()); + } + + private static IoTDBDataNodeAsyncClientManager createAsyncManager( + final List endPoints, final IntervalManager candidatePorts) { return new IoTDBDataNodeAsyncClientManager( endPoints, true, @@ -198,6 +230,7 @@ private static IoTDBDataNodeAsyncClientManager createAsyncManager( true, true, false, + candidatePorts, true); } @@ -250,6 +283,7 @@ private TestIoTDBDataNodeSyncClientManager(final List endPoints) { "sync", true, true, + new IntervalManager<>(), true); } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 0aafc35842770..0d43c04a1c98a 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -966,4 +966,9 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable generation failed: "; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "Unknown task status: "; + public static final String EXCEPTION_FAILED_TO_FIND_AN_AVAILABLE_SENDING_PORTS_IN_ARG_25FA5CA4 = "Failed to find an available sending ports in %s."; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_A_PORT_SHALL_EITHER_BE_PORT_OR_MINPORT_MAXPORT_011EDA95 = "The port %s is invalid, a port shall either be 'port' or 'minPort-maxPort'."; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_AN_INTEGER_42E9013A = "The port %s is invalid, the port must be an integer"; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_0_AND_65535_3F3581C0 = "The port %s is invalid, the port must be >= 0 and <= 65535"; + public static final String EXCEPTION_THE_PORT_ARG_ARG_IS_INVALID_THE_MIN_VALUE_SHALL_BE_SMALLER_THAN_THE_MAX_VALUE_IN_THE_PORT_RANGE_27AA8C92 = "The port %s-%s is invalid, the min value shall be smaller than the max value in the port range"; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 9746b71500d2d..147b63534cc5b 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -932,4 +932,9 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable 生成失败："; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "未知任务状态："; + public static final String EXCEPTION_FAILED_TO_FIND_AN_AVAILABLE_SENDING_PORTS_IN_ARG_25FA5CA4 = "无法在 %s 中找到可用的发送端口。"; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_A_PORT_SHALL_EITHER_BE_PORT_OR_MINPORT_MAXPORT_011EDA95 = "端口 %s 无效，端口应为 'port' 或 'minPort-maxPort'。"; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_AN_INTEGER_42E9013A = "端口 %s 无效，端口必须为整数"; + public static final String EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_0_AND_65535_3F3581C0 = "端口 %s 无效，端口必须 >= 0 且 <= 65535"; + public static final String EXCEPTION_THE_PORT_ARG_ARG_IS_INVALID_THE_MIN_VALUE_SHALL_BE_SMALLER_THAN_THE_MAX_VALUE_IN_THE_PORT_RANGE_27AA8C92 = "端口 %s-%s 无效，端口范围中的最小值应小于最大值"; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java index ce5c836688980..61e34a05c5012 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ClientPoolFactory.java @@ -39,6 +39,8 @@ import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.commons.pool2.impl.GenericKeyedObjectPool; @@ -310,6 +312,12 @@ public AsyncDataNodeMPPDataExchangeServiceClientPoolFactory( public static class AsyncPipeDataTransferServiceClientPoolFactory implements IClientPoolFactory { + final IntervalManager candidatePorts; + + public AsyncPipeDataTransferServiceClientPoolFactory( + IntervalManager candidatePorts) { + this.candidatePorts = candidatePorts; + } @Override public GenericKeyedObjectPool createClientPool( @@ -323,7 +331,8 @@ public GenericKeyedObjectPool cre .setRpcThriftCompressionEnabled(conf.isPipeSinkRPCThriftCompressionEnabled()) .setSelectorNumOfAsyncClientManager(conf.getPipeAsyncSinkSelectorNumber()) .build(), - ThreadName.PIPE_ASYNC_SINK_CLIENT_POOL.getName()), + ThreadName.PIPE_ASYNC_SINK_CLIENT_POOL.getName(), + this.candidatePorts), new ClientPoolProperty.Builder() .setMaxClientNumForEachNode(conf.getPipeAsyncSinkMaxClientNumber()) .build() @@ -336,6 +345,13 @@ public GenericKeyedObjectPool cre public static class AsyncPipeTsFileDataTransferServiceClientPoolFactory implements IClientPoolFactory { + final IntervalManager candidatePorts; + + public AsyncPipeTsFileDataTransferServiceClientPoolFactory( + IntervalManager candidatePorts) { + this.candidatePorts = candidatePorts; + } + @Override public GenericKeyedObjectPool createClientPool( ClientManager manager) { @@ -349,7 +365,8 @@ public GenericKeyedObjectPool cre .setSelectorNumOfAsyncClientManager(conf.getPipeAsyncSinkSelectorNumber()) .setPrintLogWhenEncounterException(conf.isPrintLogWhenEncounterException()) .build(), - ThreadName.PIPE_ASYNC_SINK_CLIENT_POOL.getName()), + ThreadName.PIPE_ASYNC_SINK_CLIENT_POOL.getName(), + this.candidatePorts), new ClientPoolProperty.Builder() .setMaxClientNumForEachNode(conf.getPipeAsyncSinkMaxTsFileClientNumber()) .build() diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java index 7e01b08ef2120..86d5ab645dda7 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClient.java @@ -25,6 +25,9 @@ import org.apache.iotdb.commons.client.factory.AsyncThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.i18n.ClientMessages; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; +import org.apache.iotdb.commons.pipe.sink.client.port.IoTDBSinkPortBinder; import org.apache.iotdb.rpc.TNonblockingTransportWrapper; import org.apache.iotdb.service.rpc.thrift.IClientRPCService; @@ -37,6 +40,8 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; @@ -63,13 +68,19 @@ public AsyncPipeDataTransferServiceClient( final ThriftClientProperty property, final TEndPoint endpoint, final TAsyncClientManager tClientManager, - final ClientManager clientManager) + final ClientManager clientManager, + final IntervalManager candidatePorts) throws IOException { super( property.getProtocolFactory(), tClientManager, TNonblockingTransportWrapper.wrap( endpoint.getIp(), endpoint.getPort(), property.getConnectionTimeoutMs())); + final SocketChannel socketChannel = ((TNonblockingSocket) ___transport).getSocketChannel(); + if (!___transport.isOpen()) { + IoTDBSinkPortBinder.bindPort( + candidatePorts, (sendPort) -> socketChannel.bind(new InetSocketAddress(sendPort))); + } setTimeout(property.getConnectionTimeoutMs()); this.printLogWhenEncounterException = property.isPrintLogWhenEncounterException(); this.endpoint = endpoint; @@ -212,12 +223,15 @@ public String toString() { public static class Factory extends AsyncThriftClientFactory { + final IntervalManager candidatePorts; public Factory( final ClientManager clientManager, final ThriftClientProperty thriftClientProperty, - final String threadName) { + final String threadName, + final IntervalManager candidatePorts) { super(clientManager, thriftClientProperty, threadName); + this.candidatePorts = candidatePorts; } @Override @@ -235,7 +249,8 @@ public PooledObject makeObject(final TEndPoi thriftClientProperty, endPoint, tManagers[Math.floorMod(clientCnt.incrementAndGet(), tManagers.length)], - clientManager)); + clientManager, + this.candidatePorts)); } @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java index c0c8a195f0544..865c7112113aa 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSourceConstructor; import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMetaKeeper; import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant; +import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskTemporaryRuntimeEnvironment; import org.apache.iotdb.pipe.api.PipeConnector; @@ -38,6 +39,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -137,7 +139,22 @@ protected PipeConnector validateSink( temporarySink.customize( sinkParameters, new PipeTaskRuntimeConfiguration(new PipeTaskTemporaryRuntimeEnvironment(pipeName))); - temporarySink.handshake(); + // We do not handshake when send-ports is configured because: + // 1. If handshaking without port configuration, the potentially existing firewall may fail + // the validation + // 2. The TCP will be in "TIME_WAIT" after close and will be occupied when the pipe task is + // created + // 3. When you alter pipe with the old ports, the ports are still being used by the old pipe + if (sinkParameters + .getStringOrDefault( + Arrays.asList( + PipeSinkConstant.CONNECTOR_IOTDB_SEND_PORTS_KEY, + PipeSinkConstant.SINK_IOTDB_SEND_PORTS_KEY), + PipeSinkConstant.CONNECTOR_IOTDB_SEND_PORTS_DEFAULT_VALUE) + .trim() + .isEmpty()) { + temporarySink.handshake(); + } } finally { try { temporarySink.close(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java index 04a744611c058..8ca12e7b1ddc6 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java @@ -46,6 +46,9 @@ public class PipeSinkConstant { public static final String SINK_IOTDB_PORT_KEY = "sink.port"; public static final String CONNECTOR_IOTDB_NODE_URLS_KEY = "connector.node-urls"; public static final String SINK_IOTDB_NODE_URLS_KEY = "sink.node-urls"; + public static final String CONNECTOR_IOTDB_SEND_PORTS_KEY = "connector.send-ports"; + public static final String SINK_IOTDB_SEND_PORTS_KEY = "sink.send-ports"; + public static final String CONNECTOR_IOTDB_SEND_PORTS_DEFAULT_VALUE = ""; public static final String CONNECTOR_IOTDB_SSL_ENABLE_KEY = "connector.ssl.enable"; public static final String SINK_IOTDB_SSL_ENABLE_KEY = "sink.ssl.enable"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/IntervalManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/IntervalManager.java index cecdef312de33..0a0bf588d7cec 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/IntervalManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/IntervalManager.java @@ -21,6 +21,8 @@ import javax.annotation.concurrent.NotThreadSafe; +import java.util.Iterator; +import java.util.Objects; import java.util.TreeSet; @NotThreadSafe @@ -68,10 +70,45 @@ public boolean remove(final T interval) { return false; } + public Iterator iterator() { + return new Iterator() { + private final Iterator iterator = intervals.iterator(); + private T currentInterval; + private long nextNum; + + @Override + public boolean hasNext() { + return Objects.nonNull(currentInterval) || iterator.hasNext(); + } + + @Override + public Long next() { + if (Objects.isNull(currentInterval)) { + currentInterval = iterator.next(); + nextNum = currentInterval.start; + } + final long result = nextNum++; + if (nextNum > currentInterval.end) { + if (iterator.hasNext()) { + currentInterval = iterator.next(); + nextNum = currentInterval.start; + } else { + currentInterval = null; + } + } + return result; + } + }; + } + public int size() { return intervals.size(); } + public boolean isEmpty() { + return intervals.isEmpty(); + } + @Override public String toString() { return "IntervalManager{" + "intervals=" + intervals + '}'; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/PlainInterval.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/PlainInterval.java new file mode 100644 index 0000000000000..7c4fd18194f96 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/interval/PlainInterval.java @@ -0,0 +1,26 @@ +/* + * 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.commons.pipe.datastructure.interval; + +public class PlainInterval extends Interval { + public PlainInterval(final long start, final long end) { + super(start, end); + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBClientManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBClientManager.java index cdc0cd31727a2..2251743d76640 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBClientManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBClientManager.java @@ -23,6 +23,8 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.i18n.ClientMessages; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant; import org.slf4j.Logger; @@ -61,6 +63,8 @@ public abstract class IoTDBClientManager { // it is a DataNode receiver. The flag is useless for configNode receiver. protected boolean supportModsIfIsDataNodeReceiver = true; + protected IntervalManager candidatePorts; + private static final int MAX_CONNECTION_TIMEOUT_MS = 24 * 60 * 60 * 1000; // 1 day private static final int FIRST_ADJUSTMENT_TIMEOUT_MS = 6 * 60 * 60 * 1000; // 6 hours protected static final AtomicInteger CONNECTION_TIMEOUT_MS = @@ -77,6 +81,7 @@ protected IoTDBClientManager( final String loadTsFileStrategy, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { this.endPointList = endPointList; @@ -88,6 +93,8 @@ protected IoTDBClientManager( this.loadTsFileStrategy = loadTsFileStrategy; this.validateTsFile = validateTsFile; this.shouldMarkAsPipeRequest = shouldMarkAsPipeRequest; + + this.candidatePorts = candidatePorts; this.skipIfNoPrivileges = skipIfNoPrivileges; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java index 296848629fd7e..ab285e28e549c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClient.java @@ -23,9 +23,13 @@ import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.i18n.ClientMessages; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; +import org.apache.iotdb.commons.pipe.sink.client.port.IoTDBSinkPortBinder; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqBuilder; import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import org.apache.iotdb.rpc.RpcSslUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.TimeoutChangeableTransport; import org.apache.iotdb.service.rpc.thrift.IClientRPCService; @@ -33,11 +37,21 @@ import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; import org.apache.thrift.TException; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSocket; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.security.GeneralSecurityException; + public class IoTDBSyncClient extends IClientRPCService.Client implements ThriftClient, AutoCloseable { @@ -48,12 +62,115 @@ public class IoTDBSyncClient extends IClientRPCService.Client private final TEndPoint endPoint; public IoTDBSyncClient( - ThriftClientProperty property, - String ipAddress, - int port, - boolean useSSL, - String trustStore, - String trustStorePwd) + final ThriftClientProperty property, + final String ipAddress, + final int port, + final boolean useSSL, + final String trustStore, + final String trustStorePwd, + final String keyStore, + final String keyStorePwd, + final IntervalManager candidatePorts) + throws TTransportException { + super( + createProtocol( + property, + ipAddress, + port, + useSSL, + trustStore, + trustStorePwd, + keyStore, + keyStorePwd, + candidatePorts)); + this.ipAddress = ipAddress; + this.port = port; + this.endPoint = new TEndPoint(ipAddress, port); + final TTransport transport = getInputProtocol().getTransport(); + if (!transport.isOpen()) { + transport.open(); + } + } + + private static TProtocol createProtocol( + final ThriftClientProperty property, + final String ipAddress, + final int port, + final boolean useSSL, + final String trustStore, + final String trustStorePwd, + final String keyStore, + final String keyStorePwd, + final IntervalManager candidatePorts) + throws TTransportException { + final int connectionTimeoutMs = property.getConnectionTimeoutMs(); + if (candidatePorts.isEmpty()) { + return property + .getProtocolFactory() + .getProtocol( + useSSL + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( + ipAddress, + port, + connectionTimeoutMs, + trustStore, + trustStorePwd, + keyStore, + keyStorePwd) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + ipAddress, port, connectionTimeoutMs)); + } + + final SSLContext sslContext; + try { + sslContext = + useSSL + ? RpcSslUtils.createSSLContext(keyStore, keyStorePwd, trustStore, trustStorePwd) + : null; + } catch (final GeneralSecurityException | IOException e) { + throw new TTransportException(e); + } + + final TTransport[] transport = new TTransport[1]; + IoTDBSinkPortBinder.bindPort( + candidatePorts, + (sendPort) -> { + Socket socket = new Socket(); + try { + socket.bind(new InetSocketAddress(sendPort)); + socket.connect(new InetSocketAddress(ipAddress, port), connectionTimeoutMs); + socket.setSoTimeout(connectionTimeoutMs); + if (sslContext != null) { + final SSLSocket sslSocket = + (SSLSocket) + sslContext.getSocketFactory().createSocket(socket, ipAddress, port, true); + sslSocket.setSoTimeout(connectionTimeoutMs); + sslSocket.startHandshake(); + socket = sslSocket; + } + + final TSocket thriftSocket = new TSocket(socket); + thriftSocket.setTimeout(connectionTimeoutMs); + transport[0] = DeepCopyRpcTransportFactory.INSTANCE.getTransport(thriftSocket); + } catch (final Exception e) { + try { + socket.close(); + } catch (final IOException closeException) { + e.addSuppressed(closeException); + } + throw e; + } + }); + return property.getProtocolFactory().getProtocol(transport[0]); + } + + public IoTDBSyncClient( + final ThriftClientProperty property, + final String ipAddress, + final int port, + final boolean useSSL, + final String trustStore, + final String trustStorePwd) throws TTransportException { this(property, ipAddress, port, useSSL, trustStore, trustStorePwd, null, null); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java index 8b56c30c498d6..5575a34f9f277 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientManager.java @@ -25,6 +25,8 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.i18n.PipeMessages; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferHandshakeV1Req; @@ -70,22 +72,23 @@ public abstract class IoTDBSyncClientManager extends IoTDBClientManager implemen private final LoadBalancer loadBalancer; protected IoTDBSyncClientManager( - List endPoints, - boolean useSSL, - String trustStorePath, - String trustStorePwd, - String keyStorePath, - String keyStorePwd, + final List endPoints, + final boolean useSSL, + final String trustStorePath, + final String trustStorePwd, + final String keyStorePath, + final String keyStorePwd, /* The following parameters are used locally. */ - boolean useLeaderCache, - String loadBalanceStrategy, + final boolean useLeaderCache, + final String loadBalanceStrategy, /* The following parameters are used to handshake with the receiver. */ - UserEntity userEntity, - String password, - boolean shouldReceiverConvertOnTypeMismatch, - String loadTsFileStrategy, - boolean validateTsFile, - boolean shouldMarkAsPipeRequest, + final UserEntity userEntity, + final String password, + final boolean shouldReceiverConvertOnTypeMismatch, + final String loadTsFileStrategy, + final boolean validateTsFile, + final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges) { super( endPoints, @@ -96,6 +99,7 @@ protected IoTDBSyncClientManager( loadTsFileStrategy, validateTsFile, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); this.useSSL = useSSL; @@ -219,7 +223,8 @@ private boolean initClientAndStatus( trustStorePath, trustStorePwd, keyStorePath, - keyStorePwd)); + keyStorePwd, + candidatePorts)); return true; } catch (Exception e) { endPoint2HandshakeErrorMessage.put(endPoint, e.getMessage()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/port/IoTDBSinkPortBinder.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/port/IoTDBSinkPortBinder.java new file mode 100644 index 0000000000000..eeffa4b7eb027 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/client/port/IoTDBSinkPortBinder.java @@ -0,0 +1,78 @@ +/* + * 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.commons.pipe.sink.client.port; + +import org.apache.iotdb.commons.i18n.PipeMessages; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; +import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; +import org.apache.iotdb.commons.utils.function.Consumer; +import org.apache.iotdb.pipe.api.exception.PipeConnectionException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Iterator; + +public class IoTDBSinkPortBinder { + + private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBSinkPortBinder.class); + + // ===========================bind================================ + + /** + * Find an available port in the configured ports. + * + * @param candidatePorts the ports + * @param consumer the binding process, must be called when successfully find a port + */ + public static void bindPort( + final IntervalManager candidatePorts, + final Consumer consumer) { + // Bind random port when the ports are not configured + if (candidatePorts.isEmpty()) { + try { + consumer.accept(0); + } catch (final Exception e) { + throw new PipeConnectionException(e.getMessage(), e); + } + return; + } + final Iterator iterator = candidatePorts.iterator(); + boolean portFound = false; + while (iterator.hasNext()) { + final int port = Math.toIntExact(iterator.next()); + try { + consumer.accept(port); + portFound = true; + break; + } catch (final Exception ignore) { + } + } + if (!portFound) { + final String exceptionMessage = + String.format( + PipeMessages.EXCEPTION_FAILED_TO_FIND_AN_AVAILABLE_SENDING_PORTS_IN_ARG_25FA5CA4, + candidatePorts); + PipeLogger.log(LOGGER::warn, exceptionMessage); + throw new PipeConnectionException(exceptionMessage); + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java index 6919b3b87ce72..4baa0733237da 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.i18n.PipeMessages; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; +import org.apache.iotdb.commons.pipe.sink.client.port.IoTDBSinkPortBinder; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapELanguageConstant; import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapOneByteResponse; import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils; @@ -177,12 +178,23 @@ public void handshake() throws Exception { } } - final AirGapSocket socket = new AirGapSocket(ip, port); - try { - socket.connect(new InetSocketAddress(ip, port), handshakeTimeoutMs); - socket.setKeepAlive(true); - sockets.set(i, socket); + final int finalI = i; + IoTDBSinkPortBinder.bindPort( + candidatePorts, + (sendPort) -> { + final AirGapSocket socket = new AirGapSocket(ip, port); + try { + socket.bind(new InetSocketAddress(sendPort)); + socket.connect(new InetSocketAddress(ip, port), handshakeTimeoutMs); + socket.setKeepAlive(true); + sockets.set(finalI, socket); + } catch (final Exception e) { + socket.close(); + throw e; + } + }); + LOGGER.debug(PipeMessages.CONNECTED_TO_TARGET_SERVER, ip, port); failLogTimes.remove(nodeUrls.get(i)); } catch (final Exception e) { @@ -197,7 +209,7 @@ public void handshake() throws Exception { } try { - sendHandshakeReq(socket); + sendHandshakeReq(sockets.get(i)); isSocketAlive.set(i, true); } catch (Exception e) { PipeLogger.log( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java index 601ad31a0f42e..362558218499d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java @@ -25,6 +25,8 @@ import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler; import org.apache.iotdb.commons.pipe.sink.compressor.PipeCompressor; import org.apache.iotdb.commons.pipe.sink.compressor.PipeCompressorConfig; @@ -100,6 +102,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SEND_PORTS_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SEND_PORTS_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_SKIP_IF_NO_PRIVILEGES; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_DEFAULT_VALUE; @@ -141,6 +145,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_NODE_URLS_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PORT_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_SEND_PORTS_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USERNAME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_ID; import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_KEY; @@ -169,6 +174,8 @@ public abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEvent protected UserEntity userEntity; protected String password = CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE; + protected IntervalManager candidatePorts; + protected String loadBalanceStrategy; protected String loadTsFileStrategy; @@ -281,6 +288,12 @@ public void validate(final PipeParameterValidator validator) throws Exception { Arrays.asList(CONNECTOR_IOTDB_PASSWORD_KEY, SINK_IOTDB_PASSWORD_KEY), CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE); + this.candidatePorts = + parsePorts( + parameters.getStringOrDefault( + Arrays.asList(CONNECTOR_IOTDB_SEND_PORTS_KEY, SINK_IOTDB_SEND_PORTS_KEY), + CONNECTOR_IOTDB_SEND_PORTS_DEFAULT_VALUE)); + loadBalanceStrategy = parameters .getStringOrDefault( @@ -607,6 +620,58 @@ private void checkNodeUrls(final Set nodeUrls) throws PipeParameterNo } } + private static IntervalManager parsePorts(final String ports) + throws PipeParameterNotValidException { + final IntervalManager result = new IntervalManager<>(); + if (ports == null || ports.trim().isEmpty()) { + return result; + } + String[] range; + for (String port : ports.split(",", -1)) { + range = port.trim().split("-", -1); + if (range.length > 2 + || range.length == 0 + || (range.length == 2 && (range[0].isEmpty() || range[1].isEmpty()))) { + throw new PipeParameterNotValidException( + String.format( + PipeMessages + .EXCEPTION_THE_PORT_ARG_IS_INVALID_A_PORT_SHALL_EITHER_BE_PORT_OR_MINPORT_MAXPORT_011EDA95, + port)); + } + for (final String singlePort : range) { + final long portNum; + try { + portNum = Long.parseLong(singlePort); + } catch (final Exception e) { + throw new PipeParameterNotValidException( + String.format( + PipeMessages + .EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_AN_INTEGER_42E9013A, + singlePort)); + } + if (portNum < 0 || portNum > 65535) { + throw new PipeParameterNotValidException( + String.format( + PipeMessages + .EXCEPTION_THE_PORT_ARG_IS_INVALID_THE_PORT_MUST_BE_0_AND_65535_3F3581C0, + singlePort)); + } + } + final long min = Long.parseLong(range[0]); + final long max = range.length > 1 ? Long.parseLong(range[1]) : min; + if (min > max) { + throw new PipeParameterNotValidException( + String.format( + PipeMessages + .EXCEPTION_THE_PORT_ARG_ARG_IS_INVALID_THE_MIN_VALUE_SHALL_BE_SMALLER_THAN_THE_MAX_VALUE_IN_THE_PORT_RANGE_27AA8C92, + min, + max)); + } + result.addInterval(new PlainInterval(min, max)); + } + return result; + } + @Override public void close() { // TODO: Not all the limiters should be closed here, but it's fine for now. diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java index 7ffb2ddba22d0..5dfe8e41aec86 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSslSyncSink.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.i18n.PipeMessages; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient; import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClientManager; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFilePieceReq; @@ -239,6 +241,7 @@ public void customize( loadTsFileStrategy, loadTsFileValidation, shouldMarkAsPipeRequest, + candidatePorts, skipIfNoPrivileges); clientManager.setPipeInfo(pipeName, creationTime); } @@ -260,6 +263,7 @@ protected abstract IoTDBSyncClientManager constructClient( final String loadTsFileStrategy, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest, + final IntervalManager candidatePorts, final boolean skipIfNoPrivileges); @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/function/Consumer.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/function/Consumer.java new file mode 100644 index 0000000000000..e6c84451d7907 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/function/Consumer.java @@ -0,0 +1,24 @@ +/* + * 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.commons.utils.function; + +public interface Consumer { + void accept(INPUT1 var1) throws THROWABLE; +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClientTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClientTest.java new file mode 100644 index 0000000000000..08d63f510a6bc --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/async/AsyncPipeDataTransferServiceClientTest.java @@ -0,0 +1,74 @@ +/* + * 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.commons.client.async; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.client.ClientManager; +import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; + +import org.apache.thrift.async.TAsyncClient; +import org.apache.thrift.async.TAsyncClientManager; +import org.apache.thrift.transport.TNonblockingSocket; +import org.junit.Assert; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.nio.channels.SocketChannel; + +import static org.mockito.Mockito.mock; + +public class AsyncPipeDataTransferServiceClientTest { + + @Test + public void testConfiguredSendPortBindsWithoutConnectingDuringConstruction() throws Exception { + try (final ServerSocket serverSocket = new ServerSocket(0)) { + final IntervalManager candidatePorts = new IntervalManager<>(); + candidatePorts.addInterval(new PlainInterval(0, 0)); + final TAsyncClientManager thriftClientManager = new TAsyncClientManager(); + AsyncPipeDataTransferServiceClient client = null; + try { + client = + new AsyncPipeDataTransferServiceClient( + new ThriftClientProperty.Builder().setConnectionTimeoutMs(1000).build(), + new TEndPoint("127.0.0.1", serverSocket.getLocalPort()), + thriftClientManager, + mock(ClientManager.class), + candidatePorts); + + final Field transportField = TAsyncClient.class.getDeclaredField("___transport"); + transportField.setAccessible(true); + final TNonblockingSocket nonblockingSocket = + (TNonblockingSocket) transportField.get(client); + final SocketChannel channel = nonblockingSocket.getSocketChannel(); + Assert.assertFalse(channel.isConnected()); + Assert.assertTrue(((InetSocketAddress) channel.getLocalAddress()).getPort() > 0); + } finally { + if (client != null) { + client.close(); + } + thriftClientManager.stop(); + } + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/IntervalManagerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/IntervalManagerTest.java new file mode 100644 index 0000000000000..4fcca93e9d17f --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/datastructure/IntervalManagerTest.java @@ -0,0 +1,44 @@ +/* + * 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.commons.pipe.datastructure; + +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.Iterator; + +public class IntervalManagerTest { + @Test + public void testIntervalManager() { + final IntervalManager sinkPortManager = new IntervalManager<>(); + sinkPortManager.addInterval(new PlainInterval(3, 3)); + sinkPortManager.addInterval(new PlainInterval(2, 4)); + sinkPortManager.addInterval(new PlainInterval(60000, 60000)); + final Iterator iterator = sinkPortManager.iterator(); + Assert.assertEquals(2L, (long) iterator.next()); + Assert.assertEquals(3L, (long) iterator.next()); + Assert.assertEquals(4L, (long) iterator.next()); + Assert.assertEquals(60000L, (long) iterator.next()); + Assert.assertFalse(iterator.hasNext()); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientPortTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientPortTest.java new file mode 100644 index 0000000000000..c15a0bc022fbc --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/client/IoTDBSyncClientPortTest.java @@ -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. + */ + +package org.apache.iotdb.commons.pipe.sink.client; + +import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.pipe.datastructure.interval.IntervalManager; +import org.apache.iotdb.commons.pipe.datastructure.interval.PlainInterval; +import org.apache.iotdb.rpc.TElasticFramedTransport; + +import org.apache.thrift.transport.TSocket; +import org.junit.Assert; +import org.junit.Test; + +import java.net.ServerSocket; +import java.net.Socket; + +public class IoTDBSyncClientPortTest { + + @Test + public void testDefaultSendPortsKeepsSyncConnectionOpen() throws Exception { + try (final ServerSocket receiver = new ServerSocket(0)) { + final IntervalManager candidatePorts = new IntervalManager<>(); + try (final IoTDBSyncClient client = + new IoTDBSyncClient( + new ThriftClientProperty.Builder().setConnectionTimeoutMs(1000).build(), + "127.0.0.1", + receiver.getLocalPort(), + false, + null, + null, + null, + null, + candidatePorts); + final Socket accepted = receiver.accept()) { + Assert.assertTrue(client.getInputProtocol().getTransport().isOpen()); + } + } + } + + @Test + public void testConfiguredSendPortIsUsedForSyncConnection() throws Exception { + try (final ServerSocket receiver = new ServerSocket(0); + final ServerSocket sendPortReservation = new ServerSocket(0)) { + final int sendPort = sendPortReservation.getLocalPort(); + sendPortReservation.close(); + + final IntervalManager candidatePorts = new IntervalManager<>(); + candidatePorts.addInterval(new PlainInterval(sendPort, sendPort)); + try (final IoTDBSyncClient client = + new IoTDBSyncClient( + new ThriftClientProperty.Builder().setConnectionTimeoutMs(1000).build(), + "127.0.0.1", + receiver.getLocalPort(), + false, + null, + null, + null, + null, + candidatePorts); + final Socket accepted = receiver.accept()) { + final TSocket thriftSocket = + (TSocket) + ((TElasticFramedTransport) client.getInputProtocol().getTransport()).getSocket(); + Assert.assertEquals(sendPort, thriftSocket.getSocket().getLocalPort()); + } + } + } +}