This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 3989a231 [Improve](source) Restore incremental startup from external
offsets (#708)
3989a231 is described below
commit 3989a231e6a7ad1ec4af4552da2bed10eaa07c3f
Author: wudi <[email protected]>
AuthorDate: Tue Oct 13 10:15:15 2026 +0800
[Improve](source) Restore incremental startup from external offsets (#708)
Problem Summary
A new incremental source job currently ignores offsets previously published
to source.binlog.offset-table: initial repeats its snapshot and from-timestamp
reuses the configured start time.
Read the configured consumer's saved boundary once during new Source
initialization and start directly in the incremental phase when a valid offset
exists. Restored Flink Source state remains authoritative and does not query
the external table. Without a saved record, initial performs its first
snapshot, from-timestamp uses its configured time, and latest starts from the
current Doris timestamp. Connection, permission, duplicate-record, and
invalid-timestamp failures are propagated.
This change reuses the existing offset configuration and publication path.
It does not change tablet scheduling, Reader/Enumerator coordination, or
checkpoint formats.
The S3 TVF integration fixture also uses the Doris Docker Hub MinIO mirror
to resolve the image-pull authorization failure in CI.
---
.../source/assigners/DorisSourceSplitAssigner.java | 57 ++++-
.../flink/source/reader/DorisOffsetPublisher.java | 56 ++++-
.../doris/flink/table/DorisConfigOptions.java | 6 +-
.../assigners/DorisSourceOffsetRestoreTest.java | 254 +++++++++++++++++++++
.../apache/doris/flink/sink/S3TvfSinkITCase.java | 3 +-
.../source/AbstractDorisIncrementalITCase.java | 2 +-
.../flink/source/DorisIncrementalSourceITCase.java | 59 +++++
7 files changed, 422 insertions(+), 15 deletions(-)
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
index 587f3b71..5fbc594f 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java
@@ -21,10 +21,12 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.source.DorisSourceScanMode;
import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
+import org.apache.doris.flink.source.reader.DorisOffsetPublisher;
import org.apache.doris.flink.source.split.DorisSnapshotSplit;
import org.apache.doris.flink.source.split.DorisSourceSplit;
import org.apache.doris.flink.source.split.DorisStreamSplit;
@@ -53,14 +55,40 @@ public class DorisSourceSplitAssigner {
public static DorisSourceSplitAssigner create(
DorisOptions options, DorisReadOptions readOptions, int
sourceParallelism) {
+ String externalOffset = readExternalOffset(options,
readOptions).orElse(null);
Supplier<String> currentTimestampSupplier =
() -> RestService.resolveCurrentTimestamp(options,
readOptions, LOG);
DorisSnapshotSplitAssigner snapshotSplitAssigner =
- readOptions.getScanMode().isSnapshotPhaseRequired()
+ externalOffset == null &&
readOptions.getScanMode().isSnapshotPhaseRequired()
? new DorisSnapshotSplitAssigner(options, readOptions)
: new
DorisSnapshotSplitAssigner(Collections.emptyList());
return create(
- readOptions, sourceParallelism, currentTimestampSupplier,
snapshotSplitAssigner);
+ readOptions,
+ sourceParallelism,
+ currentTimestampSupplier,
+ snapshotSplitAssigner,
+ externalOffset);
+ }
+
+ private static Optional<String> readExternalOffset(
+ DorisOptions options, DorisReadOptions readOptions) {
+ if (!readOptions.getScanMode().hasIncrementalPhase()
+ || readOptions.getBinlogOffsetTable() == null) {
+ return Optional.empty();
+ }
+ try (DorisOffsetPublisher publisher =
+ new DorisOffsetPublisher(
+ new SimpleJdbcConnectionProvider(options),
+ readOptions.getBinlogOffsetTable(),
+ readOptions.getBinlogConsumerId())) {
+ Optional<String> offset = publisher.readOffset();
+ LOG.info(
+ "Read external offset {} from {} for consumer {}.",
+ offset.orElse("<not found>"),
+ readOptions.getBinlogOffsetTable(),
+ readOptions.getBinlogConsumerId());
+ return offset;
+ }
}
public static DorisSourceSplitAssigner restore(
@@ -86,7 +114,11 @@ public class DorisSourceSplitAssigner {
? new DorisSnapshotSplitAssigner(snapshotSplits)
: new
DorisSnapshotSplitAssigner(Collections.emptyList());
return create(
- readOptions, sourceParallelism, currentTimestampSupplier,
snapshotSplitAssigner);
+ readOptions,
+ sourceParallelism,
+ currentTimestampSupplier,
+ snapshotSplitAssigner,
+ null);
}
@VisibleForTesting
@@ -102,29 +134,34 @@ public class DorisSourceSplitAssigner {
DorisReadOptions readOptions,
int sourceParallelism,
Supplier<String> currentTimestampSupplier,
- DorisSnapshotSplitAssigner snapshotSplitAssigner) {
+ DorisSnapshotSplitAssigner snapshotSplitAssigner,
+ @Nullable String externalOffset) {
DorisSourceScanMode scanMode = readOptions.getScanMode();
DorisStreamSplitAssigner streamAssigner = null;
if (scanMode.hasIncrementalPhase()) {
- String startTimestamp =
- scanMode == DorisSourceScanMode.FROM_TIMESTAMP
- ? readOptions.getScanTimestamp()
- : currentTimestampSupplier.get();
+ String startTimestamp = externalOffset;
+ if (startTimestamp == null) {
+ startTimestamp =
+ scanMode == DorisSourceScanMode.FROM_TIMESTAMP
+ ? readOptions.getScanTimestamp()
+ : currentTimestampSupplier.get();
+ }
streamAssigner =
new DorisStreamSplitAssigner(
startTimestamp, Collections.emptyList(),
currentTimestampSupplier);
}
DorisSourceCheckpoint.Phase phase =
- scanMode.isSnapshotPhaseRequired()
+ externalOffset == null && scanMode.isSnapshotPhaseRequired()
? DorisSourceCheckpoint.Phase.SNAPSHOT
: DorisSourceCheckpoint.Phase.STREAM;
DorisSourceSplitAssigner assigner =
new DorisSourceSplitAssigner(
scanMode, sourceParallelism, phase,
snapshotSplitAssigner, streamAssigner);
LOG.info(
- "Initialized Doris source assigner in {} mode at phase {} with
{} pending snapshot splits.",
+ "Initialized Doris source assigner in {} mode at phase {} with
stream start timestamp {} and {} pending snapshot splits.",
scanMode,
phase,
+ assigner.nextStreamStartTimestamp(),
snapshotSplitAssigner.remainingSplits().size());
return assigner;
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
index 76d6f351..38018db7 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/reader/DorisOffsetPublisher.java
@@ -18,14 +18,20 @@ package org.apache.doris.flink.source.reader;
import org.apache.doris.flink.connection.JdbcConnectionProvider;
import org.apache.doris.flink.exception.DorisRuntimeException;
+import org.apache.doris.flink.source.split.DorisStreamSplit;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.time.format.ResolverStyle;
+import java.util.Optional;
import java.util.function.Consumer;
-/** Publishes completed-checkpoint offsets through Doris JDBC. */
+/** Reads startup offsets and publishes completed-checkpoint offsets through
Doris JDBC. */
public class DorisOffsetPublisher implements AutoCloseable {
private final JdbcConnectionProvider connectionProvider;
private final String database;
@@ -70,6 +76,54 @@ public class DorisOffsetPublisher implements AutoCloseable {
throw new DorisRuntimeException("Offset table does not exist: " +
database + "." + table);
}
+ /** Returns empty only when this consumer has no saved offset. */
+ public Optional<String> readOffset() {
+ String selectSql =
+ "SELECT `offset_timestamp` FROM "
+ + quoteIdentifier(database)
+ + "."
+ + quoteIdentifier(table)
+ + " WHERE `consumer_id` = ?";
+ try (PreparedStatement statement =
+
connectionProvider.getOrEstablishConnection().prepareStatement(selectSql)) {
+ statement.setString(1, consumerId);
+ try (ResultSet result = statement.executeQuery()) {
+ if (!result.next()) {
+ return Optional.empty();
+ }
+ String offset = result.getString("offset_timestamp");
+ if (result.next()) {
+ throw new DorisRuntimeException("Multiple offsets for
consumer " + consumerId);
+ }
+ String invalidOffset =
+ "Invalid offset timestamp for consumer " + consumerId
+ ": " + offset;
+ if (!DorisStreamSplit.isValidTimestamp(offset)) {
+ throw new DorisRuntimeException(invalidOffset);
+ }
+ try {
+ LocalDateTime.parse(
+ offset,
+ DateTimeFormatter.ofPattern("uuuu-MM-dd HH:mm:ss")
+ .withResolverStyle(ResolverStyle.STRICT));
+ } catch (DateTimeParseException error) {
+ throw new DorisRuntimeException(invalidOffset, error);
+ }
+ return Optional.of(offset);
+ }
+ } catch (DorisRuntimeException error) {
+ throw error;
+ } catch (Exception error) {
+ throw new DorisRuntimeException(
+ "Failed to read offset from "
+ + database
+ + "."
+ + table
+ + " for consumer "
+ + consumerId,
+ error);
+ }
+ }
+
/** Invokes the callback with null on success or the publication error on
failure. */
public void publish(String offset, Consumer<Exception> callback) {
Exception failure = null;
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
index 3b159679..3cd66764 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/table/DorisConfigOptions.java
@@ -419,7 +419,8 @@ public class DorisConfigOptions {
ConfigOptions.key("source.scan.timestamp")
.stringType()
.noDefaultValue()
- .withDescription("Inclusive start timestamp for
from-timestamp mode");
+ .withDescription(
+ "Inclusive first-start timestamp for
from-timestamp mode; a saved external offset takes precedence");
public static final ConfigOption<String> SOURCE_BINLOG_INCREMENT_TYPE =
ConfigOptions.key("source.binlog.increment-type")
.stringType()
@@ -443,7 +444,8 @@ public class DorisConfigOptions {
ConfigOptions.key("source.binlog.offset-table")
.stringType()
.noDefaultValue()
- .withDescription("Doris table used to persist completed
checkpoint offsets");
+ .withDescription(
+ "Doris table used to persist completed checkpoint
offsets and resume new jobs without restored Source state");
public static final ConfigOption<String> SOURCE_BINLOG_CONSUMER_ID =
ConfigOptions.key("source.binlog.consumer-id")
.stringType()
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
new file mode 100644
index 00000000..dcf39914
--- /dev/null
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java
@@ -0,0 +1,254 @@
+// 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.doris.flink.source.assigners;
+
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
+import org.apache.doris.flink.rest.RestService;
+import org.apache.doris.flink.source.DorisSourceScanMode;
+import org.apache.doris.flink.source.enumerator.DorisSourceCheckpoint;
+import org.apache.doris.flink.source.split.DorisStreamSplit;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class DorisSourceOffsetRestoreTest {
+ private static final String SAVED = "2026-10-09 12:00:00";
+ private static final String CONFIGURED = "2026-10-09 10:00:00";
+ private static final String NOW = "2026-10-09 13:00:00";
+ private final DorisOptions options =
+ DorisOptions.builder()
+ .setFenodes("localhost:8030")
+ .setTableIdentifier("db.t")
+ .setJdbcUrl("jdbc:mysql://localhost:9030")
+ .build();
+ private final Connection connection = mock(Connection.class);
+ private final PreparedStatement statement = mock(PreparedStatement.class);
+ private final ResultSet result = mock(ResultSet.class);
+
+ @BeforeEach
+ void setUp() throws Exception {
+ when(connection.prepareStatement(any())).thenReturn(statement);
+ when(statement.executeQuery()).thenReturn(result);
+ when(result.next()).thenReturn(true, false);
+ when(result.getString("offset_timestamp")).thenReturn(SAVED);
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = DorisSourceScanMode.class,
+ names = {"INITIAL", "LATEST", "FROM_TIMESTAMP"})
+ void resumesEveryIncrementalModeFromSavedBoundary(DorisSourceScanMode
mode) throws Exception {
+ DorisReadOptions readOptions = readOptions(mode);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> providers =
providers();
+ MockedStatic<RestService> rest = rest(readOptions)) {
+ DorisSourceSplitAssigner assigner =
+ DorisSourceSplitAssigner.create(options, readOptions, 2);
+
assertThat(assigner.getPhase()).isEqualTo(DorisSourceCheckpoint.Phase.STREAM);
+ assertThat(assigner.hasPendingSnapshotSplits()).isFalse();
+ assertThat(assigner.getScanMode()).isEqualTo(mode);
+
assertThat(assigner.snapshotState(0L).getNextStreamStartTimestamp()).isEqualTo(SAVED);
+
assigner.onDiscoveredEndTimestamp(assigner.discoverNextStreamEndTimestamp());
+ DorisStreamSplit window = (DorisStreamSplit)
assigner.getNext().get();
+ assertThat(window.getStartTimestamp()).isEqualTo(SAVED);
+ assertThat(window.getEndTimestamp()).isEqualTo(NOW);
+ rest.verify(
+ () -> RestService.findPartitions(eq(options),
eq(readOptions), any()), never());
+ verify(connection)
+ .prepareStatement(
+ "SELECT `offset_timestamp` FROM `ops`.`offsets`
WHERE `consumer_id` = ?");
+ verify(statement).setString(1, "orders-consumer");
+ verify(result).close();
+ verify(statement).close();
+ verify(providers.constructed().get(0)).closeConnection();
+ }
+ }
+
+ @Test
+ void checkpointRestorationDoesNotOpenOffsetConnection() {
+ DorisReadOptions readOptions =
readOptions(DorisSourceScanMode.INITIAL);
+ DorisSourceCheckpoint checkpoint =
+ new DorisSourceCheckpoint(
+ DorisSourceCheckpoint.Phase.STREAM, SAVED, 2,
Collections.emptyList());
+ try (MockedConstruction<SimpleJdbcConnectionProvider> providers =
providers()) {
+ DorisSourceSplitAssigner assigner =
+ DorisSourceSplitAssigner.restore(options, readOptions,
checkpoint, 2);
+
assertThat(assigner.snapshotState(0L).getNextStreamStartTimestamp()).isEqualTo(SAVED);
+
assertThat(assigner.getPhase()).isEqualTo(DorisSourceCheckpoint.Phase.STREAM);
+ assertThat(providers.constructed()).isEmpty();
+ }
+ }
+
+ @Test
+ void initialWithoutRecordPlansFirstSnapshot() throws Exception {
+ when(result.next()).thenReturn(false);
+ DorisReadOptions readOptions =
readOptions(DorisSourceScanMode.INITIAL);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
providers();
+ MockedStatic<RestService> rest = rest(readOptions)) {
+ DorisSourceSplitAssigner assigner =
+ DorisSourceSplitAssigner.create(options, readOptions, 2);
+
assertThat(assigner.getPhase()).isEqualTo(DorisSourceCheckpoint.Phase.SNAPSHOT);
+
assertThat(assigner.snapshotState(0L).getNextStreamStartTimestamp()).isEqualTo(NOW);
+ rest.verify(() -> RestService.findPartitions(eq(options),
eq(readOptions), any()));
+ }
+ }
+
+ @Test
+ void fromTimestampWithoutRecordUsesConfiguredStart() throws Exception {
+ when(result.next()).thenReturn(false);
+ DorisReadOptions readOptions =
readOptions(DorisSourceScanMode.FROM_TIMESTAMP);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
providers()) {
+ assertThat(
+ DorisSourceSplitAssigner.create(options,
readOptions, 2)
+ .snapshotState(0L)
+ .getNextStreamStartTimestamp())
+ .isEqualTo(CONFIGURED);
+ }
+ }
+
+ @Test
+ void latestWithoutRecordStartsFromCurrentTime() throws Exception {
+ when(result.next()).thenReturn(false);
+ DorisReadOptions readOptions = readOptions(DorisSourceScanMode.LATEST);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
providers();
+ MockedStatic<RestService> rest = rest(readOptions)) {
+ DorisSourceSplitAssigner assigner =
+ DorisSourceSplitAssigner.create(options, readOptions, 2);
+
assertThat(assigner.getPhase()).isEqualTo(DorisSourceCheckpoint.Phase.STREAM);
+ assertThat(assigner.hasPendingSnapshotSplits()).isFalse();
+
assertThat(assigner.snapshotState(0L).getNextStreamStartTimestamp()).isEqualTo(NOW);
+ rest.verify(
+ () -> RestService.findPartitions(eq(options),
eq(readOptions), any()), never());
+ }
+ }
+
+ @Test
+ void latestWithoutOffsetTableKeepsCurrentTimeStart() {
+ DorisReadOptions readOptions =
+
DorisReadOptions.builder().setScanMode(DorisSourceScanMode.LATEST).build();
+ try (MockedConstruction<SimpleJdbcConnectionProvider> providers =
providers();
+ MockedStatic<RestService> ignored = rest(readOptions)) {
+ assertThat(
+ DorisSourceSplitAssigner.create(options,
readOptions, 2)
+ .snapshotState(0L)
+ .getNextStreamStartTimestamp())
+ .isEqualTo(NOW);
+ assertThat(providers.constructed()).isEmpty();
+ }
+ }
+
+ @Test
+ void duplicateOffsetsFailInsteadOfChoosingOne() throws Exception {
+ when(result.next()).thenReturn(true, true);
+ assertStartupFails(DorisSourceScanMode.INITIAL, "Multiple offsets");
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(
+ strings = {
+ "2026-02-30 12:00:00",
+ "2026-10-09 25:00:00",
+ "2026-10-09 12:00:00.123",
+ "not-a-timestamp"
+ })
+ void invalidSavedOffsetsFailInsteadOfStartingSnapshot(String value) throws
Exception {
+ when(result.getString("offset_timestamp")).thenReturn(value);
+ assertStartupFails(DorisSourceScanMode.INITIAL, "Invalid offset
timestamp");
+ }
+
+ @Test
+ void queryFailureDoesNotFallBackToConfiguredTimestamp() throws Exception {
+ when(statement.executeQuery()).thenThrow(new SQLException("SELECT
denied"));
+ assertStartupFails(DorisSourceScanMode.FROM_TIMESTAMP, "Failed to read
offset");
+ }
+
+ @Test
+ void connectionFailureDoesNotFallBackToSnapshot() throws Exception {
+ DorisReadOptions readOptions =
readOptions(DorisSourceScanMode.INITIAL);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
+ mockConstruction(
+ SimpleJdbcConnectionProvider.class,
+ (provider, context) ->
+
when(provider.getOrEstablishConnection())
+ .thenThrow(new
SQLException("unavailable")));
+ MockedStatic<RestService> rest = rest(readOptions)) {
+ assertThatThrownBy(() -> DorisSourceSplitAssigner.create(options,
readOptions, 2))
+ .hasMessageContaining("Failed to read offset");
+ }
+ }
+
+ private void assertStartupFails(DorisSourceScanMode mode, String message) {
+ DorisReadOptions readOptions = readOptions(mode);
+ try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
providers();
+ MockedStatic<RestService> rest = rest(readOptions)) {
+ assertThatThrownBy(() -> DorisSourceSplitAssigner.create(options,
readOptions, 2))
+ .hasMessageContaining(message);
+ }
+ }
+
+ private MockedConstruction<SimpleJdbcConnectionProvider> providers() {
+ return mockConstruction(
+ SimpleJdbcConnectionProvider.class,
+ (provider, context) ->
+
when(provider.getOrEstablishConnection()).thenReturn(connection));
+ }
+
+ private MockedStatic<RestService> rest(DorisReadOptions readOptions) {
+ MockedStatic<RestService> rest = mockStatic(RestService.class);
+ rest.when(() -> RestService.resolveCurrentTimestamp(eq(options),
eq(readOptions), any()))
+ .thenReturn(NOW);
+ rest.when(() -> RestService.parseIdentifier(eq("db.t"), any()))
+ .thenReturn(new String[] {"db", "t"});
+ rest.when(() -> RestService.findPartitions(eq(options),
eq(readOptions), any()))
+ .thenReturn(Collections.emptyList());
+ return rest;
+ }
+
+ private DorisReadOptions readOptions(DorisSourceScanMode mode) {
+ return DorisReadOptions.builder()
+ .setScanMode(mode)
+ .setScanTimestamp(mode == DorisSourceScanMode.FROM_TIMESTAMP ?
CONFIGURED : null)
+ .setBinlogOffsetTable("ops.offsets")
+ .setBinlogConsumerId("orders-consumer")
+ .build();
+ }
+}
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/S3TvfSinkITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/S3TvfSinkITCase.java
index 58858404..b8854936 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/S3TvfSinkITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/sink/S3TvfSinkITCase.java
@@ -95,7 +95,8 @@ public class S3TvfSinkITCase extends AbstractITCaseService {
private static final String ACCESS_KEY = "minioadmin";
private static final String SECRET_KEY = "minioadmin";
private static final int MINIO_PORT = 9000;
- private static final String MINIO_IMAGE =
"quay.io/minio/minio:RELEASE.2024-10-13T13-34-11Z";
+ private static final String MINIO_IMAGE =
+ "doristhirdpartydocker/minio:RELEASE.2024-11-07T00-52-20Z";
private static GenericContainer<?> minio;
private static S3Client s3Client;
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
index cc1ae1b3..4f7c470c 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/AbstractDorisIncrementalITCase.java
@@ -238,7 +238,7 @@ public abstract class AbstractDorisIncrementalITCase
extends AbstractITCaseServi
return collector.containsAll(expected);
},
Deadline.fromNow(DEFAULT_TIMEOUT),
- 100L,
+ 1000L,
"Missing source rows " + expected + "; observed=" +
collector.getRows());
}
diff --git
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
index 548602a2..565d54c6 100644
---
a/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
+++
b/flink-doris-connector/flink-doris-connector-it/src/test/java/org/apache/doris/flink/source/DorisIncrementalSourceITCase.java
@@ -167,6 +167,65 @@ public class DorisIncrementalSourceITCase extends
AbstractDorisIncrementalITCase
}
}
+ @Test
+ public void testFromTimestampRestartsFromExternalOffsetWithoutFlinkState()
throws Exception {
+ restartFromExternalOffset("from-timestamp");
+ }
+
+ @Test
+ public void testInitialRestartsWithoutRepeatingSnapshot() throws Exception
{
+ restartFromExternalOffset("initial");
+ }
+
+ private void restartFromExternalOffset(String scanMode) throws Exception {
+ String table = "tbl_external_offset_restart_" + scanMode.replace('-',
'_');
+ initializeIncrementalTable(table, "(0,'historical','ignored')");
+ initializeOffsetTable();
+ String configuredStart = resolveCurrentDorisTimestamp(table);
+ waitForDorisTimestampAfter(table, configuredStart);
+ String savedOffset;
+ try (IncrementalResultCollector first =
+ startSourceWithOffsetPersistence(
+ table,
+ scanMode,
+ "from-timestamp".equals(scanMode) ? configuredStart :
null,
+ OFFSET_TABLE,
+ CONSUMER_ID)) {
+ executeDorisSql(
+ String.format(
+ "INSERT INTO %s.%s VALUES
(1,'before-stop','ignored')",
+ DATABASE, table));
+ checkpointUntilContains(
+ first, Arrays.asList(ObservedRow.of(RowKind.INSERT, 1,
"before-stop")));
+ String writtenAt = resolveCurrentDorisTimestamp(table);
+ waitForDorisTimestampAfter(table, writtenAt);
+ savedOffset = checkpointUntilOffsetAfter(first, writtenAt);
+ }
+
+ // These writes must be recovered by a new job with no
checkpoint/savepoint restore path.
+ executeDorisSql(
+ String.format(
+ "INSERT INTO %s.%s VALUES
(2,'while-stopped-a','ignored'),(3,'while-stopped-b','ignored')",
+ DATABASE, table));
+ waitForDorisTimestampAfter(table, savedOffset);
+ try (IncrementalResultCollector restarted =
+ startSourceWithOffsetPersistence(
+ table,
+ scanMode,
+ "from-timestamp".equals(scanMode) ? configuredStart :
null,
+ OFFSET_TABLE,
+ CONSUMER_ID)) {
+ checkpointUntilContains(
+ restarted,
+ Arrays.asList(
+ ObservedRow.of(RowKind.INSERT, 2,
"while-stopped-a"),
+ ObservedRow.of(RowKind.INSERT, 3,
"while-stopped-b")));
+ Assert.assertFalse(
+ "Restart replayed rows before the saved boundary: " +
restarted.getRows(),
+ restarted.getRows().stream().anyMatch(row -> row.getInt(0)
< 2));
+ }
+ }
+
private void initializeOffsetTable() {
executeDorisSql(
String.format("DROP TABLE IF EXISTS %s.%s", DATABASE,
OFFSET_TABLE),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
