This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12741-2f3bcd276a8a2ae4fc177e25c6c635d3eb527ee7 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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)); + } +}
