This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new ff58662570 [Fix][Connector-V2] Fix InfluxDB sink retention policy,
null timestamp fallback and flush retry loop (#12741)
ff58662570 is described below
commit ff586625703eac77000109ecb9e2a646af655e36
Author: Jast <[email protected]>
AuthorDate: Sat Oct 17 06:58:52 2026 +0000
[Fix][Connector-V2] Fix InfluxDB sink retention policy, null timestamp
fallback and flush retry loop (#12741)
---
seatunnel-connectors-v2/connector-influxdb/pom.xml | 5 +
.../seatunnel/influxdb/client/InfluxDBClient.java | 2 +-
.../influxdb/serialize/DefaultSerializer.java | 2 +
.../influxdb/sink/InfluxDBSinkWriter.java | 10 +-
.../influxdb/serialize/DefaultSerializerTest.java | 122 ++++++++++++++++++
.../influxdb/sink/InfluxDBSinkWriterTest.java | 136 +++++++++++++++++++++
6 files changed, 274 insertions(+), 3 deletions(-)
diff --git a/seatunnel-connectors-v2/connector-influxdb/pom.xml
b/seatunnel-connectors-v2/connector-influxdb/pom.xml
index 2631cadb8b..c6fc74fbb9 100644
--- a/seatunnel-connectors-v2/connector-influxdb/pom.xml
+++ b/seatunnel-connectors-v2/connector-influxdb/pom.xml
@@ -49,6 +49,11 @@
<artifactId>influxdb-java</artifactId>
<version>${influxdb.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/client/InfluxDBClient.java
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/client/InfluxDBClient.java
index b09ed09dfe..32641ec033 100644
---
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/client/InfluxDBClient.java
+++
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/client/InfluxDBClient.java
@@ -92,7 +92,7 @@ public class InfluxDBClient {
public static InfluxDB getWriteClient(SinkConfig sinkConfig) throws
ConnectException {
InfluxDB influxdb = getInfluxDB(sinkConfig);
influxdb.setDatabase(sinkConfig.getDatabase());
- setWriteProperty(getInfluxDB(sinkConfig), sinkConfig);
+ setWriteProperty(influxdb, sinkConfig);
return influxdb;
}
}
diff --git
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializer.java
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializer.java
index 4ec7004215..2a63e4103d 100644
---
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializer.java
+++
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializer.java
@@ -122,7 +122,9 @@ public class DefaultSerializer implements Serializer {
return (row, builder) -> {
Object time = row.getField(timeFieldIndex);
if (time == null) {
+ // use processing time when the value of the configured time
column is null
builder.time(System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
+ return;
}
SeaTunnelDataType<?> timestampFieldType =
seaTunnelRowType.getFieldType(timeFieldIndex);
switch (timestampFieldType.getSqlType()) {
diff --git
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java
index b0d23c7e79..f940621f57 100644
---
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriter.java
@@ -108,10 +108,12 @@ public class InfluxDBSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
return;
}
BatchPoints.Builder batchPoints =
BatchPoints.database(sinkConfig.getDatabase());
+ // The builder appends points on every call, so populate it once
before the retry loop
+ batchPoints.points(batchList);
for (int i = 0; i <= sinkConfig.getMaxRetries(); i++) {
try {
- batchPoints.points(batchList);
influxdb.write(batchPoints.build());
+ break;
} catch (Exception e) {
log.error("Writing records to influxdb failed, retry times =
{}", i, e);
if (i >= sinkConfig.getMaxRetries()) {
@@ -151,7 +153,7 @@ public class InfluxDBSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
public void connect() throws ConnectException {
if (influxdb == null) {
- influxdb = InfluxDBClient.getWriteClient(sinkConfig);
+ influxdb = createClient();
String version = influxdb.version();
if (!influxdb.ping().isGood()) {
throw new InfluxdbConnectorException(
@@ -163,4 +165,8 @@ public class InfluxDBSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
log.info("connect influxdb successful. sever version :{}.",
version);
}
}
+
+ protected InfluxDB createClient() throws ConnectException {
+ return InfluxDBClient.getWriteClient(sinkConfig);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializerTest.java
b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializerTest.java
new file mode 100644
index 0000000000..2da6ff6cfb
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/serialize/DefaultSerializerTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.influxdb.serialize;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.influxdb.dto.Point;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+public class DefaultSerializerTest {
+
+ private static final String MEASUREMENT = "cpu";
+
+ private DefaultSerializer serializer(SeaTunnelDataType<?> timeType) {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"time", "value"},
+ new SeaTunnelDataType[] {timeType,
BasicType.DOUBLE_TYPE});
+ return new DefaultSerializer(
+ rowType, TimeUnit.NANOSECONDS, Collections.emptyList(),
"time", MEASUREMENT);
+ }
+
+ @Test
+ public void shouldFallBackToCurrentTimeWhenStringTimeColumnIsNull() {
+ DefaultSerializer serializer = serializer(BasicType.STRING_TYPE);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{null, 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertNotNull(point.lineProtocol());
+ }
+
+ @Test
+ public void shouldFallBackToCurrentTimeWhenBigintTimeColumnIsNull() {
+ DefaultSerializer serializer = serializer(BasicType.LONG_TYPE);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{null, 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertNotNull(point.lineProtocol());
+ }
+
+ @Test
+ public void shouldFallBackToCurrentTimeWhenTimestampTimeColumnIsNull() {
+ DefaultSerializer serializer =
serializer(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{null, 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertNotNull(point.lineProtocol());
+ }
+
+ @Test
+ public void shouldUseConfiguredTimeColumnWhenPresent() {
+ DefaultSerializer serializer = serializer(BasicType.STRING_TYPE);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{"123456", 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertTrue(point.lineProtocol().contains("123456"));
+ }
+
+ @Test
+ public void shouldSerializeTags() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"host", "time", "value"},
+ new SeaTunnelDataType[] {
+ BasicType.STRING_TYPE, BasicType.LONG_TYPE,
BasicType.DOUBLE_TYPE
+ });
+ DefaultSerializer serializer =
+ new DefaultSerializer(
+ rowType,
+ TimeUnit.NANOSECONDS,
+ Collections.singletonList("host"),
+ "time",
+ MEASUREMENT);
+ Point point =
+ serializer.serialize(new SeaTunnelRow(new Object[]
{"server-1", 123456L, 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertTrue(point.lineProtocol().contains("server-1"));
+ Assertions.assertTrue(point.lineProtocol().contains("123456"));
+ }
+
+ @Test
+ public void shouldUseProcessingTimeWhenTimeKeyNotConfigured() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"value"}, new SeaTunnelDataType[]
{BasicType.DOUBLE_TYPE});
+ DefaultSerializer serializer =
+ new DefaultSerializer(
+ rowType, TimeUnit.NANOSECONDS,
Collections.emptyList(), null, MEASUREMENT);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertNotNull(point.lineProtocol());
+ }
+
+ @Test
+ public void shouldSerializeLocalDateTimeTimeColumn() {
+ DefaultSerializer serializer =
serializer(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ LocalDateTime time = LocalDateTime.of(2024, 1, 1, 0, 0, 0);
+ Point point = serializer.serialize(new SeaTunnelRow(new Object[]
{time, 1.0d}));
+ Assertions.assertNotNull(point);
+ Assertions.assertNotNull(point.lineProtocol());
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriterTest.java
b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriterTest.java
new file mode 100644
index 0000000000..36191f7520
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-influxdb/src/test/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkWriterTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.influxdb.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.influxdb.config.SinkConfig;
+import
org.apache.seatunnel.connectors.seatunnel.influxdb.exception.InfluxdbConnectorException;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Pong;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class InfluxDBSinkWriterTest {
+
+ private static SinkConfig sinkConfig(int maxRetries) {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("url", "http://127.0.0.1:8086";);
+ SinkConfig sinkConfig = new
SinkConfig(ReadonlyConfig.fromMap(configMap));
+ sinkConfig.setBatchSize(100);
+ sinkConfig.setMaxRetries(maxRetries);
+ sinkConfig.setRetryBackoffMultiplierMs(0);
+ sinkConfig.setMaxRetryBackoffMs(0);
+ return sinkConfig;
+ }
+
+ private static InfluxDBSinkWriter writerWithClient(InfluxDB client,
SinkConfig sinkConfig)
+ throws Exception {
+ return new InfluxDBSinkWriter(
+ sinkConfig,
+ new SeaTunnelRowType(
+ new String[] {"time", "value"},
+ new SeaTunnelDataType[] {BasicType.LONG_TYPE,
BasicType.DOUBLE_TYPE})) {
+ @Override
+ protected InfluxDB createClient() {
+ return client;
+ }
+ };
+ }
+
+ private static InfluxDB healthyClient() {
+ InfluxDB client = mock(InfluxDB.class);
+ when(client.version()).thenReturn("1.8.0");
+ Pong pong = new Pong();
+ pong.setVersion("1.8.0");
+ when(client.ping()).thenReturn(pong);
+ return client;
+ }
+
+ private static Point point(long time) {
+ return Point.measurement("cpu")
+ .time(time, TimeUnit.MILLISECONDS)
+ .addField("value", 1.0d)
+ .build();
+ }
+
+ @Test
+ public void shouldWriteBatchExactlyOnceWhenFirstAttemptSucceeds() throws
Exception {
+ InfluxDB client = healthyClient();
+ InfluxDBSinkWriter writer = writerWithClient(client, sinkConfig(3));
+
+ writer.write(point(1L));
+ writer.write(point(2L));
+ writer.flush();
+
+ ArgumentCaptor<BatchPoints> captor =
ArgumentCaptor.forClass(BatchPoints.class);
+ verify(client, times(1)).write(captor.capture());
+ Assertions.assertEquals(2, captor.getValue().getPoints().size());
+ }
+
+ @Test
+ public void shouldNotAccumulatePointsAcrossRetryAttempts() throws
Exception {
+ InfluxDB client = healthyClient();
+ doThrow(new RuntimeException("first attempt fails"))
+ .doAnswer(invocation -> null)
+ .when(client)
+ .write(any(BatchPoints.class));
+ InfluxDBSinkWriter writer = writerWithClient(client, sinkConfig(3));
+
+ writer.write(point(1L));
+ writer.write(point(2L));
+ writer.flush();
+
+ ArgumentCaptor<BatchPoints> captor =
ArgumentCaptor.forClass(BatchPoints.class);
+ verify(client, times(2)).write(captor.capture());
+ for (BatchPoints batch : captor.getAllValues()) {
+ Assertions.assertEquals(2, batch.getPoints().size());
+ }
+ }
+
+ @Test
+ public void shouldThrowAfterRetriesExhausted() throws Exception {
+ InfluxDB client = healthyClient();
+ doThrow(new RuntimeException("influxdb unavailable"))
+ .when(client)
+ .write(any(BatchPoints.class));
+ InfluxDBSinkWriter writer = writerWithClient(client, sinkConfig(2));
+
+ writer.write(point(1L));
+ Assertions.assertThrows(InfluxdbConnectorException.class,
writer::flush);
+
+ verify(client, times(3)).write(any(BatchPoints.class));
+ }
+}
