[improve] iotdb options (#8965)

Author: liunaijie

seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java

@@ -181,9 +181,7 @@ private Set<String> buildWhiteList() {
         whiteList.add("TDengineSourceOptions");
         whiteList.add("PulsarSourceOptions");
         whiteList.add("MongodbSinkOptions");
-        whiteList.add("IoTDBSinkOptions");
         whiteList.add("PaimonSourceOptions");
-        whiteList.add("IoTDBSourceOptions");
         whiteList.add("SlsSourceOptions");
         whiteList.add("SentrySinkOptions");
         whiteList.add("QdrantSinkOptions");

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/CommonConfig.java

@@ -17,9 +17,6 @@
 
 package org.apache.seatunnel.connectors.seatunnel.iotdb.config;
 
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
-
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.ToString;
@@ -31,13 +28,6 @@
 @AllArgsConstructor
 public class CommonConfig {
 
-    public static final Option<String> NODE_URLS =
-            Options.key("node_urls").stringType().noDefaultValue().withDescription("node urls");
-    public static final Option<String> USERNAME =
-            Options.key("username").stringType().noDefaultValue().withDescription("username");
-    public static final Option<String> PASSWORD =
-            Options.key("password").stringType().noDefaultValue().withDescription("password");
-
     private final List<String> nodeUrls;
     private final String username;
     private final String password;

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/IoTDBCommonOptions.java

@@ -0,0 +1,31 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.iotdb.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class IoTDBCommonOptions {
+
+    public static final Option<String> NODE_URLS =
+            Options.key("node_urls").stringType().noDefaultValue().withDescription("node urls");
+    public static final Option<String> USERNAME =
+            Options.key("username").stringType().noDefaultValue().withDescription("username");
+    public static final Option<String> PASSWORD =
+            Options.key("password").stringType().noDefaultValue().withDescription("password");
+}

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/IoTDBSinkOptions.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.seatunnel.connectors.seatunnel.iotdb.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+import java.util.List;
+
+public class IoTDBSinkOptions extends IoTDBCommonOptions {
+
+    private static final int DEFAULT_BATCH_SIZE = 1024;
+
+    public static final Option<String> KEY_TIMESTAMP =
+            Options.key("key_timestamp")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("key timestamp");
+    public static final Option<String> KEY_DEVICE =
+            Options.key("key_device").stringType().noDefaultValue().withDescription("key device");
+    public static final Option<List<String>> KEY_MEASUREMENT_FIELDS =
+            Options.key("key_measurement_fields")
+                    .listType()
+                    .noDefaultValue()
+                    .withDescription("key measurement fields");
+    public static final Option<String> STORAGE_GROUP =
+            Options.key("storage_group")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("store group");
+    public static final Option<Integer> BATCH_SIZE =
+            Options.key("batch_size")
+                    .intType()
+                    .defaultValue(DEFAULT_BATCH_SIZE)
+                    .withDescription("batch size");
+    public static final Option<Integer> MAX_RETRIES =
+            Options.key("max_retries").intType().noDefaultValue().withDescription("max retries");
+    public static final Option<Integer> RETRY_BACKOFF_MULTIPLIER_MS =
+            Options.key("retry_backoff_multiplier_ms")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("retry backoff multiplier ms ");
+    public static final Option<Integer> MAX_RETRY_BACKOFF_MS =
+            Options.key("max_retry_backoff_ms")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("max retry backoff ms ");
+    public static final Option<Integer> DEFAULT_THRIFT_BUFFER_SIZE =
+            Options.key("default_thrift_buffer_size")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("default thrift buffer size");
+    public static final Option<Integer> MAX_THRIFT_FRAME_SIZE =
+            Options.key("max_thrift_frame_size")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("max thrift frame size");
+    public static final Option<String> ZONE_ID =
+            Options.key("zone_id").stringType().noDefaultValue().withDescription("zone id");
+    public static final Option<Boolean> ENABLE_RPC_COMPRESSION =
+            Options.key("enable_rpc_compression")
+                    .booleanType()
+                    .noDefaultValue()
+                    .withDescription("enable rpc comm");
+    public static final Option<Integer> CONNECTION_TIMEOUT_IN_MS =
+            Options.key("connection_timeout_in_ms")
+                    .intType()
+                    .noDefaultValue()
+                    .withDescription("connection timeout ms");
+}

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/SourceConfig.java renamed to seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/IoTDBSourceOptions.java

@@ -26,7 +26,7 @@
  * <p>please see the following link for more details:
  * https://iotdb.apache.org/UserGuide/Master/API/Programming-Java-Native-API.html
  */
-public class SourceConfig {
+public class IoTDBSourceOptions extends IoTDBCommonOptions {
 
     public static final Option<String> SQL =
             Options.key("sql").stringType().noDefaultValue().withDescription("sql");

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/config/SinkConfig.java

@@ -17,10 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.iotdb.config;
 
-import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
-import org.apache.seatunnel.api.configuration.Option;
-import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 
 import lombok.Getter;
 import lombok.NonNull;
@@ -30,78 +27,18 @@
 import java.time.ZoneId;
 import java.util.List;
 
-import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
 
 @Setter
 @Getter
 @ToString
 public class SinkConfig extends CommonConfig {
 
-    private static final int DEFAULT_BATCH_SIZE = 1024;
-
-    public static final Option<String> KEY_TIMESTAMP =
-            Options.key("key_timestamp")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("key timestamp");
-    public static final Option<String> KEY_DEVICE =
-            Options.key("key_device").stringType().noDefaultValue().withDescription("key device");
-    public static final Option<List<String>> KEY_MEASUREMENT_FIELDS =
-            Options.key("key_measurement_fields")
-                    .listType()
-                    .noDefaultValue()
-                    .withDescription("key measurement fields");
-    public static final Option<String> STORAGE_GROUP =
-            Options.key("storage_group")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("store group");
-    public static final Option<Integer> BATCH_SIZE =
-            Options.key("batch_size")
-                    .intType()
-                    .defaultValue(DEFAULT_BATCH_SIZE)
-                    .withDescription("batch size");
-    public static final Option<Integer> MAX_RETRIES =
-            Options.key("max_retries").intType().noDefaultValue().withDescription("max retries");
-    public static final Option<Integer> RETRY_BACKOFF_MULTIPLIER_MS =
-            Options.key("retry_backoff_multiplier_ms")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("retry backoff multiplier ms ");
-    public static final Option<Integer> MAX_RETRY_BACKOFF_MS =
-            Options.key("max_retry_backoff_ms")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("max retry backoff ms ");
-    public static final Option<Integer> DEFAULT_THRIFT_BUFFER_SIZE =
-            Options.key("default_thrift_buffer_size")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("default thrift buffer size");
-    public static final Option<Integer> MAX_THRIFT_FRAME_SIZE =
-            Options.key("max_thrift_frame_size")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("max thrift frame size");
-    public static final Option<String> ZONE_ID =
-            Options.key("zone_id").stringType().noDefaultValue().withDescription("zone id");
-    public static final Option<Boolean> ENABLE_RPC_COMPRESSION =
-            Options.key("enable_rpc_compression")
-                    .booleanType()
-                    .noDefaultValue()
-                    .withDescription("enable rpc comm");
-    public static final Option<Integer> CONNECTION_TIMEOUT_IN_MS =
-            Options.key("connection_timeout_in_ms")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("connection timeout ms");
-
     private String keyTimestamp;
     private String keyDevice;
     private List<String> keyMeasurementFields;
     private String storageGroup;
-    private int batchSize = BATCH_SIZE.defaultValue();
+    private int batchSize;
     private int maxRetries;
     private int retryBackoffMultiplierMs;
     private int maxRetryBackoffMs;
@@ -116,70 +53,50 @@ public SinkConfig(
         super(nodeUrls, username, password);
     }
 
-    public static SinkConfig loadConfig(Config pluginConfig) {
+    public static SinkConfig loadConfig(ReadonlyConfig pluginConfig) {
         SinkConfig sinkConfig =
                 new SinkConfig(
-                        pluginConfig.getStringList(NODE_URLS.key()),
-                        pluginConfig.getString(USERNAME.key()),
-                        pluginConfig.getString(PASSWORD.key()));
+                        pluginConfig.toConfig().getStringList(IoTDBSinkOptions.NODE_URLS.key()),
+                        pluginConfig.get(IoTDBSinkOptions.USERNAME),
+                        pluginConfig.get(IoTDBSinkOptions.PASSWORD));
 
-        sinkConfig.setKeyDevice(pluginConfig.getString(KEY_DEVICE.key()));
-        if (pluginConfig.hasPath(KEY_TIMESTAMP.key())) {
-            sinkConfig.setKeyTimestamp(pluginConfig.getString(KEY_TIMESTAMP.key()));
-        }
-        if (pluginConfig.hasPath(KEY_MEASUREMENT_FIELDS.key())) {
-            sinkConfig.setKeyMeasurementFields(
-                    pluginConfig.getStringList(KEY_MEASUREMENT_FIELDS.key()));
-        }
-        if (pluginConfig.hasPath(STORAGE_GROUP.key())) {
-            sinkConfig.setStorageGroup(pluginConfig.getString(STORAGE_GROUP.key()));
+        sinkConfig.setKeyDevice(pluginConfig.get(IoTDBSinkOptions.KEY_DEVICE));
+        sinkConfig.setKeyTimestamp(pluginConfig.get(IoTDBSinkOptions.KEY_TIMESTAMP));
+        sinkConfig.setKeyMeasurementFields(
+                pluginConfig.get(IoTDBSinkOptions.KEY_MEASUREMENT_FIELDS));
+        sinkConfig.setStorageGroup(pluginConfig.get(IoTDBSinkOptions.STORAGE_GROUP));
+        if (pluginConfig.getOptional(IoTDBSinkOptions.BATCH_SIZE).isPresent()) {
+            sinkConfig.setBatchSize(pluginConfig.get(IoTDBSinkOptions.BATCH_SIZE));
         }
-        if (pluginConfig.hasPath(BATCH_SIZE.key())) {
-            int batchSize = checkIntArgument(pluginConfig.getInt(BATCH_SIZE.key()));
-            sinkConfig.setBatchSize(batchSize);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.MAX_RETRIES).isPresent()) {
+            sinkConfig.setMaxRetries(pluginConfig.get(IoTDBSinkOptions.MAX_RETRIES));
         }
-        if (pluginConfig.hasPath(MAX_RETRIES.key())) {
-            int maxRetries = checkIntArgument(pluginConfig.getInt(MAX_RETRIES.key()));
-            sinkConfig.setMaxRetries(maxRetries);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.RETRY_BACKOFF_MULTIPLIER_MS).isPresent()) {
+            sinkConfig.setRetryBackoffMultiplierMs(
+                    pluginConfig.get(IoTDBSinkOptions.RETRY_BACKOFF_MULTIPLIER_MS));
         }
-        if (pluginConfig.hasPath(RETRY_BACKOFF_MULTIPLIER_MS.key())) {
-            int retryBackoffMultiplierMs =
-                    checkIntArgument(pluginConfig.getInt(RETRY_BACKOFF_MULTIPLIER_MS.key()));
-            sinkConfig.setRetryBackoffMultiplierMs(retryBackoffMultiplierMs);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.MAX_RETRY_BACKOFF_MS).isPresent()) {
+            sinkConfig.setMaxRetryBackoffMs(
+                    pluginConfig.get(IoTDBSinkOptions.MAX_RETRY_BACKOFF_MS));
         }
-        if (pluginConfig.hasPath(MAX_RETRY_BACKOFF_MS.key())) {
-            int maxRetryBackoffMs =
-                    checkIntArgument(pluginConfig.getInt(MAX_RETRY_BACKOFF_MS.key()));
-            sinkConfig.setMaxRetryBackoffMs(maxRetryBackoffMs);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.DEFAULT_THRIFT_BUFFER_SIZE).isPresent()) {
+            sinkConfig.setThriftDefaultBufferSize(
+                    pluginConfig.get(IoTDBSinkOptions.DEFAULT_THRIFT_BUFFER_SIZE));
         }
-        if (pluginConfig.hasPath(DEFAULT_THRIFT_BUFFER_SIZE.key())) {
-            int thriftDefaultBufferSize =
-                    checkIntArgument(pluginConfig.getInt(DEFAULT_THRIFT_BUFFER_SIZE.key()));
-            sinkConfig.setThriftDefaultBufferSize(thriftDefaultBufferSize);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.MAX_THRIFT_FRAME_SIZE).isPresent()) {
+            sinkConfig.setThriftMaxFrameSize(
+                    pluginConfig.get(IoTDBSinkOptions.MAX_THRIFT_FRAME_SIZE));
         }
-        if (pluginConfig.hasPath(MAX_THRIFT_FRAME_SIZE.key())) {
-            int thriftMaxFrameSize =
-                    checkIntArgument(pluginConfig.getInt(MAX_THRIFT_FRAME_SIZE.key()));
-            sinkConfig.setThriftMaxFrameSize(thriftMaxFrameSize);
+        if (pluginConfig.getOptional(IoTDBSinkOptions.ZONE_ID).isPresent()) {
+            sinkConfig.setZoneId(ZoneId.of(pluginConfig.get(IoTDBSinkOptions.ZONE_ID)));
         }
-        if (pluginConfig.hasPath(ZONE_ID.key())) {
-            sinkConfig.setZoneId(ZoneId.of(pluginConfig.getString(ZONE_ID.key())));
-        }
-        if (pluginConfig.hasPath(ENABLE_RPC_COMPRESSION.key())) {
-            sinkConfig.setEnableRPCCompression(
-                    pluginConfig.getBoolean(ENABLE_RPC_COMPRESSION.key()));
-        }
-        if (pluginConfig.hasPath(CONNECTION_TIMEOUT_IN_MS.key())) {
-            int connectionTimeoutInMs =
-                    checkIntArgument(pluginConfig.getInt(CONNECTION_TIMEOUT_IN_MS.key()));
+        sinkConfig.setEnableRPCCompression(
+                pluginConfig.get(IoTDBSinkOptions.ENABLE_RPC_COMPRESSION));
+        if (pluginConfig.getOptional(IoTDBSinkOptions.CONNECTION_TIMEOUT_IN_MS).isPresent()) {
             checkNotNull(sinkConfig.getEnableRPCCompression());
-            sinkConfig.setConnectionTimeoutInMs(connectionTimeoutInMs);
+            sinkConfig.setConnectionTimeoutInMs(
+                    pluginConfig.get(IoTDBSinkOptions.CONNECTION_TIMEOUT_IN_MS));
         }
         return sinkConfig;
     }
-
-    private static int checkIntArgument(int args) {
-        checkArgument(args > 0);
-        return args;
-    }
 }