[Improve] [Connector-V2] Remove scheduler in InfluxDB sink (apache#5271)

--------- Co-authored-by: gdliu3 <[email protected]>
ruanwenjun · Aug 24, 2023 · f459f50 · f459f50
1 parent 8bbda25
commit f459f50
Show file tree

Hide file tree

Showing 4 changed files with 1 addition and 52 deletions.
diff --git a/docs/en/connector-v2/sink/InfluxDB.md b/docs/en/connector-v2/sink/InfluxDB.md
@@ -22,7 +22,6 @@ Write data to InfluxDB.
 | key_time                    | string | no       | processing time              |
 | key_tags                    | array  | no       | exclude `field` & `key_time` |
 | batch_size                  | int    | no       | 1024                         |
-| batch_interval_ms           | int    | no       | -                            |
 | max_retries                 | int    | no       | -                            |
 | retry_backoff_multiplier_ms | int    | no       | -                            |
 | connect_timeout_ms          | long   | no       | 15000                        |
@@ -63,11 +62,7 @@ If not specified, include all fields with `influxDB` measurement field
 
 ### 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 influxDB
-
-### 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 influxDB
+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 influxDB
 
 ### max_retries [int]
 

diff --git a/...b/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/config/SinkConfig.java b/...b/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/config/SinkConfig.java
@@ -60,12 +60,6 @@ public SinkConfig(Config config) {
                     .defaultValue(1024)
                     .withDescription("batch size of the influxdb client");
 
-    public static final Option<Integer> BATCH_INTERVAL_MS =
-            Options.key("batch_interval_ms")
-                    .intType()
-                    .noDefaultValue()
-                    .withDescription("batch interval ms of the influxdb client");
-
     public static final Option<Integer> MAX_RETRIES =
             Options.key("max_retries")
                     .intType()
@@ -104,7 +98,6 @@ public SinkConfig(Config config) {
     private String keyTime;
     private List<String> keyTags;
     private int batchSize = BATCH_SIZE.defaultValue();
-    private Integer batchIntervalMs;
     private int maxRetries;
     private int retryBackoffMultiplierMs;
     private int maxRetryBackoffMs;
@@ -119,9 +112,6 @@ public static SinkConfig loadConfig(Config config) {
         if (config.hasPath(KEY_TAGS.key())) {
             sinkConfig.setKeyTags(config.getStringList(KEY_TAGS.key()));
         }
-        if (config.hasPath(BATCH_INTERVAL_MS.key())) {
-            sinkConfig.setBatchIntervalMs(config.getInt(BATCH_INTERVAL_MS.key()));
-        }
         if (config.hasPath(MAX_RETRIES.key())) {
             sinkConfig.setMaxRetries(config.getInt(MAX_RETRIES.key()));
         }

diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkFactory.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkFactory.java
@@ -28,7 +28,6 @@
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.InfluxDBConfig.PASSWORD;
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.InfluxDBConfig.URL;
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.InfluxDBConfig.USERNAME;
-import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.SinkConfig.BATCH_INTERVAL_MS;
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.SinkConfig.BATCH_SIZE;
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.SinkConfig.KEY_MEASUREMENT;
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.SinkConfig.KEY_TAGS;
@@ -54,7 +53,6 @@ public OptionRule optionRule() {
                         KEY_TAGS,
                         KEY_TIME,
                         BATCH_SIZE,
-                        BATCH_INTERVAL_MS,
                         MAX_RETRIES,
                         RETRY_BACKOFF_MULTIPLIER_MS)
                 .build();

diff --git a/...main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java b/...main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java
@@ -34,7 +34,6 @@
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
@@ -43,10 +42,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class InfluxDBSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
@@ -55,15 +50,11 @@ public class InfluxDBSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
     private InfluxDB influxdb;
     private final SinkConfig sinkConfig;
     private final List<Point> batchList;
-    private ScheduledExecutorService scheduler;
-    private ScheduledFuture<?> scheduledFuture;
     private volatile Exception flushException;
-    private final Integer batchIntervalMs;
 
     public InfluxDBSinkWriter(Config pluginConfig, SeaTunnelRowType seaTunnelRowType)
             throws ConnectException {
         this.sinkConfig = SinkConfig.loadConfig(pluginConfig);
-        this.batchIntervalMs = sinkConfig.getBatchIntervalMs();
         this.serializer =
                 new DefaultSerializer(
                         seaTunnelRowType,
@@ -73,26 +64,6 @@ public InfluxDBSinkWriter(Config pluginConfig, SeaTunnelRowType seaTunnelRowType
                         sinkConfig.getMeasurement());
         this.batchList = new ArrayList<>();
 
-        if (batchIntervalMs != null) {
-            scheduler =
-                    Executors.newSingleThreadScheduledExecutor(
-                            new ThreadFactoryBuilder()
-                                    .setNameFormat("influxDB-sink-output-%s")
-                                    .build());
-            scheduledFuture =
-                    scheduler.scheduleAtFixedRate(
-                            () -> {
-                                try {
-                                    flush();
-                                } catch (IOException e) {
-                                    flushException = e;
-                                }
-                            },
-                            batchIntervalMs,
-                            batchIntervalMs,
-                            TimeUnit.MILLISECONDS);
-        }
-
         connect();
     }
 
@@ -112,11 +83,6 @@ public Optional<Void> prepareCommit() {
 
     @Override
     public void close() throws IOException {
-        if (scheduledFuture != null) {
-            scheduledFuture.cancel(false);
-            scheduler.shutdown();
-        }
-
         flush();
 
         if (influxdb != null) {