Caideyipi commented on code in PR #13593:
URL: https://github.com/apache/iotdb/pull/13593#discussion_r1777007761
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java:
##########
@@ -175,6 +203,100 @@ public void validate(final PipeParameterValidator
validator) throws Exception {
Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY,
SINK_IOTDB_BATCH_SIZE_KEY),
CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE));
+ this.customSendPortStrategy =
+ parameters
+ .getStringOrDefault(
+ Arrays.asList(
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY,
+ SINK_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_RANGE_STRATEGY)
+ .trim()
+ .toLowerCase();
+
+ validator.validate(
+ arg ->
CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET.contains(customSendPortStrategy),
+ String.format(
+ "send port restriction strategy should be one of %s, but got %s.",
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET,
customSendPortStrategy),
+ customSendPortStrategy);
+
+ final int parallelTaskNum =
+ parameters.getIntOrDefault(
+ Arrays.asList(
+ PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
+ PipeConnectorConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
+
PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
+ final int requiredPortsNum =
+ parallelTaskNum
+ *
CommonDescriptor.getInstance().getConfig().getPipeAsyncConnectorMaxClientNumber();
+
+ if
(CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_RANGE_STRATEGY.equals(customSendPortStrategy))
{
+ minSendPortRange =
+ parameters.getIntOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SEND_PORT_MIN_KEY,
SINK_IOTDB_SEND_PORT_MIN_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_MIN_VALUE);
+ maxSendPortRange =
+ parameters.getIntOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SEND_PORT_MAX_KEY,
SINK_IOTDB_SEND_PORT_MAX_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_MAX_VALUE);
+ validator.validate(
+ args -> (int) args[0] < (int) args[1],
+ String.format(
+ "%s must be less than or equal to %s, but got %d > %d.",
+ SINK_IOTDB_SEND_PORT_MIN_KEY,
+ SINK_IOTDB_SEND_PORT_MAX_KEY,
+ minSendPortRange,
+ maxSendPortRange),
+ minSendPortRange,
+ maxSendPortRange);
+ validator.validate(
+ args -> (int) args[0] <= (int) args[1] || (int) args[2] >= (int)
args[3],
Review Comment:
Why ||?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java:
##########
@@ -175,6 +203,100 @@ public void validate(final PipeParameterValidator
validator) throws Exception {
Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY,
SINK_IOTDB_BATCH_SIZE_KEY),
CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE));
+ this.customSendPortStrategy =
+ parameters
+ .getStringOrDefault(
+ Arrays.asList(
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY,
+ SINK_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_RANGE_STRATEGY)
+ .trim()
+ .toLowerCase();
+
+ validator.validate(
+ arg ->
CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET.contains(customSendPortStrategy),
+ String.format(
+ "send port restriction strategy should be one of %s, but got %s.",
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET,
customSendPortStrategy),
+ customSendPortStrategy);
+
+ final int parallelTaskNum =
+ parameters.getIntOrDefault(
+ Arrays.asList(
+ PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
+ PipeConnectorConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
+
PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
+ final int requiredPortsNum =
+ parallelTaskNum
+ *
CommonDescriptor.getInstance().getConfig().getPipeAsyncConnectorMaxClientNumber();
Review Comment:
Is it OK to use "asyncConnector"'s max client number to detect
"requiredPortsNum"?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java:
##########
@@ -175,6 +203,100 @@ public void validate(final PipeParameterValidator
validator) throws Exception {
Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY,
SINK_IOTDB_BATCH_SIZE_KEY),
CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE));
+ this.customSendPortStrategy =
+ parameters
+ .getStringOrDefault(
+ Arrays.asList(
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY,
+ SINK_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_RANGE_STRATEGY)
+ .trim()
+ .toLowerCase();
+
+ validator.validate(
+ arg ->
CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET.contains(customSendPortStrategy),
+ String.format(
+ "send port restriction strategy should be one of %s, but got %s.",
+ CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_STRATEGY_SET,
customSendPortStrategy),
+ customSendPortStrategy);
+
+ final int parallelTaskNum =
+ parameters.getIntOrDefault(
+ Arrays.asList(
+ PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
+ PipeConnectorConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
+
PipeConnectorConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
+ final int requiredPortsNum =
+ parallelTaskNum
+ *
CommonDescriptor.getInstance().getConfig().getPipeAsyncConnectorMaxClientNumber();
+
+ if
(CONNECTOR_IOTDB_SEND_PORT_RESTRICTION_RANGE_STRATEGY.equals(customSendPortStrategy))
{
+ minSendPortRange =
+ parameters.getIntOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SEND_PORT_MIN_KEY,
SINK_IOTDB_SEND_PORT_MIN_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_MIN_VALUE);
+ maxSendPortRange =
+ parameters.getIntOrDefault(
+ Arrays.asList(CONNECTOR_IOTDB_SEND_PORT_MAX_KEY,
SINK_IOTDB_SEND_PORT_MAX_KEY),
+ CONNECTOR_IOTDB_SEND_PORT_MAX_VALUE);
+ validator.validate(
+ args -> (int) args[0] < (int) args[1],
Review Comment:
It seems that the range is [min, max] thus args[0] can be <= args[1]?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
