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 73dada03 [fix](source) Capture stream start before planning snapshot
tablets (#720)
73dada03 is described below
commit 73dada0380e359ed9516ddff0f1802e35f446298
Author: wudi <[email protected]>
AuthorDate: Fri Nov 6 10:24:34 2026 +0800
[fix](source) Capture stream start before planning snapshot tablets (#720)
---
.../source/assigners/DorisSourceSplitAssigner.java | 16 ++++++++++++----
.../source/assigners/DorisSourceOffsetRestoreTest.java | 18 +++++++++++++++++-
2 files changed, 29 insertions(+), 5 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 b33e6691..87f0ef13 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
@@ -60,6 +60,12 @@ public class DorisSourceSplitAssigner {
String externalOffset = readExternalOffset(options,
readOptions).orElse(null);
Supplier<String> currentTimestampSupplier =
() -> RestService.resolveCurrentTimestamp(options,
readOptions, LOG);
+ String streamStartTimestamp = externalOffset;
+ if (streamStartTimestamp == null
+ && readOptions.getScanMode() == DorisSourceScanMode.INITIAL) {
+ // Capture the start before planning so newly created tablets
remain in range.
+ streamStartTimestamp = currentTimestampSupplier.get();
+ }
DorisSnapshotSplitAssigner snapshotSplitAssigner =
externalOffset == null &&
readOptions.getScanMode().isSnapshotPhaseRequired()
? new DorisSnapshotSplitAssigner(options, readOptions)
@@ -70,6 +76,7 @@ public class DorisSourceSplitAssigner {
currentTimestampSupplier,
snapshotSplitAssigner,
externalOffset,
+ streamStartTimestamp,
tabletSupplier(options, readOptions));
}
@@ -124,6 +131,7 @@ public class DorisSourceSplitAssigner {
currentTimestampSupplier,
snapshotSplitAssigner,
null,
+ null,
tabletSupplier);
}
@@ -148,20 +156,20 @@ public class DorisSourceSplitAssigner {
Supplier<String> currentTimestampSupplier,
DorisSnapshotSplitAssigner snapshotSplitAssigner,
@Nullable String externalOffset,
+ @Nullable String streamStartTimestamp,
BiFunction<String, String, Collection<Long>> tabletSupplier) {
DorisSourceScanMode scanMode = readOptions.getScanMode();
DorisStreamSplitAssigner streamAssigner = null;
if (scanMode.hasIncrementalPhase()) {
- String startTimestamp = externalOffset;
- if (startTimestamp == null) {
- startTimestamp =
+ if (streamStartTimestamp == null) {
+ streamStartTimestamp =
scanMode == DorisSourceScanMode.FROM_TIMESTAMP
? readOptions.getScanTimestamp()
: currentTimestampSupplier.get();
}
streamAssigner =
new DorisStreamSplitAssigner(
- startTimestamp,
+ streamStartTimestamp,
Collections.emptyList(),
currentTimestampSupplier,
tabletSupplier,
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
index 7dbbe601..2b8fa917 100644
---
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
@@ -38,6 +38,7 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -118,16 +119,31 @@ class DorisSourceOffsetRestoreTest {
}
@Test
- void initialWithoutRecordPlansFirstSnapshot() throws Exception {
+ void initialWithoutRecordCapturesStartBeforePlanningSnapshot() throws
Exception {
when(result.next()).thenReturn(false);
DorisReadOptions readOptions =
readOptions(DorisSourceScanMode.INITIAL);
+ AtomicReference<String> currentTimestamp = new AtomicReference<>(NOW);
try (MockedConstruction<SimpleJdbcConnectionProvider> ignored =
providers();
MockedStatic<RestService> rest = rest(readOptions)) {
+ rest.when(
+ () ->
+ RestService.resolveCurrentTimestamp(
+ eq(options), eq(readOptions),
any()))
+ .thenAnswer(invocation -> currentTimestamp.get());
+ rest.when(() -> RestService.findPartitions(eq(options),
eq(readOptions), any()))
+ .thenAnswer(
+ invocation -> {
+ currentTimestamp.set("2026-10-09 13:00:10");
+ return Collections.emptyList();
+ });
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()));
+ assigner.enterStreamPhase();
+ assertThat(assigner.discoverNextStreamEndTimestamp().f0)
+ .isEqualTo("2026-10-09 13:00:10");
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
