[Improve] [Connector-V2] Remove scheduler in IoTDB sink (#5270)

---------

Co-authored-by: gdliu3 <gdliu3@iflytek.com>

Author: liugddx

docs/en/connector-v2/sink/IoTDB.md

@@ -31,7 +31,6 @@ the same `key` and `timestamp`, the new data will overwrite the old one.
 | key_measurement_fields      | array   | no       | exclude `device` & `timestamp` |
 | storage_group               | string  | no       | -                              |
 | batch_size                  | int     | no       | 1024                           |
-| batch_interval_ms           | int     | no       | -                              |
 | max_retries                 | int     | no       | -                              |
 | retry_backoff_multiplier_ms | int     | no       | -                              |
 | max_retry_backoff_ms        | int     | no       | -                              |
@@ -74,11 +73,7 @@ example: deviceId = ${storage_group} + "." +  ${key_device}
 
 ### batch_size [int]
 
-For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`, the data will be flushed into the IoTDB
-
-### batch_interval_ms [int]
-
-For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`, the data will be flushed into the IoTDB
+For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `checkpoint.interval`, the data will be flushed into the IoTDB
 
 ### max_retries [int]
 
@@ -129,7 +124,6 @@ sink {
     username = "root"
     password = "root"
     batch_size = 1024
-    batch_interval_ms = 1000
   }
 }
 ```

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

@@ -62,11 +62,6 @@ public class SinkConfig extends CommonConfig {
                     .intType()
                     .defaultValue(DEFAULT_BATCH_SIZE)
                     .withDescription("batch size");
-    public static final Option<String> BATCH_INTERVAL_MS =
-            Options.key("batch_interval_ms")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("batch interval ms");
     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 =
@@ -107,7 +102,6 @@ public class SinkConfig extends CommonConfig {
     private List<String> keyMeasurementFields;
     private String storageGroup;
     private int batchSize = BATCH_SIZE.defaultValue();
-    private Integer batchIntervalMs;
     private int maxRetries;
     private int retryBackoffMultiplierMs;
     private int maxRetryBackoffMs;
@@ -144,10 +138,6 @@ public static SinkConfig loadConfig(Config pluginConfig) {
             int batchSize = checkIntArgument(pluginConfig.getInt(BATCH_SIZE.key()));
             sinkConfig.setBatchSize(batchSize);
         }
-        if (pluginConfig.hasPath(BATCH_INTERVAL_MS.key())) {
-            int batchIntervalMs = checkIntArgument(pluginConfig.getInt(BATCH_INTERVAL_MS.key()));
-            sinkConfig.setBatchIntervalMs(batchIntervalMs);
-        }
         if (pluginConfig.hasPath(MAX_RETRIES.key())) {
             int maxRetries = checkIntArgument(pluginConfig.getInt(MAX_RETRIES.key()));
             sinkConfig.setMaxRetries(maxRetries);

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/sink/IoTDBSinkClient.java

@@ -28,17 +28,12 @@
 import org.apache.iotdb.session.Session;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class IoTDBSinkClient {
@@ -47,8 +42,6 @@ public class IoTDBSinkClient {
     private final List<IoTDBRecord> batchList;
 
     private Session session;
-    private ScheduledExecutorService scheduler;
-    private ScheduledFuture<?> scheduledFuture;
     private volatile boolean initialize;
     private volatile Exception flushException;
 
@@ -95,26 +88,6 @@ private void tryInit() throws IOException {
                     "Initialize IoTDB client failed.",
                     e);
         }
-
-        if (sinkConfig.getBatchIntervalMs() != null) {
-            scheduler =
-                    Executors.newSingleThreadScheduledExecutor(
-                            new ThreadFactoryBuilder()
-                                    .setNameFormat("IoTDB-sink-output-%s")
-                                    .build());
-            scheduledFuture =
-                    scheduler.scheduleAtFixedRate(
-                            () -> {
-                                try {
-                                    flush();
-                                } catch (IOException e) {
-                                    flushException = e;
-                                }
-                            },
-                            sinkConfig.getBatchIntervalMs(),
-                            sinkConfig.getBatchIntervalMs(),
-                            TimeUnit.MILLISECONDS);
-        }
         initialize = true;
     }
 
@@ -129,11 +102,6 @@ public synchronized void write(IoTDBRecord record) throws IOException {
     }
 
     public synchronized void close() throws IOException {
-        if (scheduledFuture != null) {
-            scheduledFuture.cancel(false);
-            scheduler.shutdown();
-        }
-
         flush();
 
         try {

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/sink/IoTDBSinkFactory.java

@@ -26,7 +26,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.CommonConfig.NODE_URLS;
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.CommonConfig.PASSWORD;
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.CommonConfig.USERNAME;
-import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.SinkConfig.BATCH_INTERVAL_MS;
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.SinkConfig.BATCH_SIZE;
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.SinkConfig.CONNECTION_TIMEOUT_IN_MS;
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.SinkConfig.DEFAULT_THRIFT_BUFFER_SIZE;
@@ -57,7 +56,6 @@ public OptionRule optionRule() {
                         KEY_MEASUREMENT_FIELDS,
                         STORAGE_GROUP,
                         BATCH_SIZE,
-                        BATCH_INTERVAL_MS,
                         MAX_RETRIES,
                         RETRY_BACKOFF_MULTIPLIER_MS,
                         MAX_RETRY_BACKOFF_MS,