From d331fd90b4bf04ad53a605a9c1b61a84d5a86607 Mon Sep 17 00:00:00 2001 From: pengfeiwwu Date: Thu, 24 Aug 2023 16:59:02 +0800 Subject: [PATCH] [BAHIR-325] Optimize influxdb sink write consistency --- .../connectors/influxdb/sink2/writer/InfluxDBWriter.java | 1 + 1 file changed, 1 insertion(+) diff --git a/flink-connector-influxdb2/src/main/java/org/apache/flink/streaming/connectors/influxdb/sink2/writer/InfluxDBWriter.java b/flink-connector-influxdb2/src/main/java/org/apache/flink/streaming/connectors/influxdb/sink2/writer/InfluxDBWriter.java index 10797ecf..299de4b0 100644 --- a/flink-connector-influxdb2/src/main/java/org/apache/flink/streaming/connectors/influxdb/sink2/writer/InfluxDBWriter.java +++ b/flink-connector-influxdb2/src/main/java/org/apache/flink/streaming/connectors/influxdb/sink2/writer/InfluxDBWriter.java @@ -89,6 +89,7 @@ public void write(IN in, Context context) throws IOException, InterruptedExcepti public void flush(boolean flush) throws IOException, InterruptedException { if (this.lastTimestamp == 0) return; + this.writeCurrentElements(); commit(Collections.singletonList(this.lastTimestamp)); }