Copilot commented on code in PR #720:
URL:
https://github.com/apache/doris-flink-connector/pull/720#discussion_r4479646988
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/source/assigners/DorisSourceSplitAssigner.java:
##########
@@ -148,11 +155,11 @@ private static DorisSourceSplitAssigner create(
Supplier<String> currentTimestampSupplier,
DorisSnapshotSplitAssigner snapshotSplitAssigner,
@Nullable String externalOffset,
+ @Nullable String startTimestamp,
BiFunction<String, String, Collection<Long>> tabletSupplier) {
Review Comment:
`startTimestamp` is ambiguous next to `externalOffset` and doesn’t convey
that it’s specifically the *stream start timestamp captured for the incremental
phase* (potentially pre-snapshot planning). Consider renaming it to something
like `streamStartTimestamp` or `initialStreamStartTimestamp` (and rename
corresponding locals/call sites) to make the separation of concerns
self-evident.
##########
flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/source/assigners/DorisSourceOffsetRestoreTest.java:
##########
@@ -118,16 +119,31 @@ void checkpointRestorationDoesNotOpenOffsetConnection() {
}
@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");
Review Comment:
This regression test currently encodes ordering via a side-effect in the
`findPartitions` stub plus a hard-coded later timestamp string. To make the
intent more explicit and less brittle, consider (a) using Mockito `InOrder` to
assert `resolveCurrentTimestamp(...)` is invoked before `findPartitions(...)`
during assigner creation, and (b) extracting the later timestamp into a named
constant (or deriving it from `NOW`) so the test communicates the timeline
without magic string literals.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
