Use long windows when catching up

scylla-cdc-base/src/main/java/com/scylladb/cdc/model/worker/TaskAction.java

@@ -9,6 +9,7 @@
 import com.google.common.flogger.FluentLogger;
 import com.scylladb.cdc.cql.WorkerCQL.Reader;
 import com.scylladb.cdc.model.FutureUtils;
+import com.scylladb.cdc.model.Timestamp;
 
 abstract class TaskAction {
     private static final FluentLogger logger = FluentLogger.forEnclosingClass();
@@ -180,7 +181,9 @@ public MoveToNextWindowTaskAction(WorkerConfiguration workerConfiguration, Task
 
         @Override
         public CompletableFuture<TaskAction> run() {
-            TaskState newState = task.state.moveToNextWindow(workerConfiguration.queryTimeWindowSizeMs);
+            Date now = Date.from(workerConfiguration.getClock().instant());
+
+            TaskState newState = task.state.moveToNextWindow(new Timestamp(now), workerConfiguration.confidenceWindowSizeMs, workerConfiguration.queryTimeWindowSizeMs);
             workerConfiguration.transport.moveStateToNextWindow(task.id, newState);
             Task newTask = task.updateState(newState);
             return CompletableFuture.completedFuture(new ReadNewWindowTaskAction(workerConfiguration, newTask, 0));

scylla-cdc-base/src/main/java/com/scylladb/cdc/model/worker/TaskState.java

@@ -54,8 +54,16 @@ public Timestamp getWindowEndTimestamp() {
         return windowEnd;
     }
 
-    public TaskState moveToNextWindow(long nextWindowSizeMs) {
-        return new TaskState(windowEnd, windowEnd.plus(nextWindowSizeMs, ChronoUnit.MILLIS), Optional.empty());
+    public TaskState moveToNextWindow(Timestamp now, long confidenceWindowSizeMs, long newQueryWindowSizeMs) {
+        Timestamp newWindowEnd = now.plus(-confidenceWindowSizeMs, ChronoUnit.MILLIS);
+
+        // Make sure that the window is at least newQueryWindowSizeMs long.
+        long windowLength = ChronoUnit.MILLIS.between(windowEnd.toDate().toInstant(), newWindowEnd.toDate().toInstant());
+        if (windowLength < newQueryWindowSizeMs) {
+            newWindowEnd = windowEnd.plus(newQueryWindowSizeMs, ChronoUnit.MILLIS);
+        }
+
+        return new TaskState(windowEnd, newWindowEnd, Optional.empty());
     }
 
     public TaskState update(ChangeId seen) {
@@ -86,12 +94,24 @@ public String toString() {
     /*
      * Creates an initial state for tasks in a given |generation|.
      *
-     * Such initial state starts at the beginning of the generation and spans for
-     * |windowSizeMs| milliseconds.
+     * Such initial state starts at the beginning of the generation and spans
+     * until now minus confidence window size.
      */
-    public static TaskState createInitialFor(GenerationId generation, long windowSizeMs) {
-        Timestamp generationStart = generation.getGenerationStart();
-        return new TaskState(generationStart, generationStart.plus(windowSizeMs, ChronoUnit.MILLIS), Optional.empty());
+    public static TaskState createInitialFor(GenerationId generation, Timestamp now,
+                                             long confidenceWindowSizeMs, long queryTimeWindowSizeMs) {
+        // Start reading at generation start:
+        Timestamp windowStart = generation.getGenerationStart();
+
+        // Create a large window up to (now - confidenceWindowSizeMs), except
+        // when the window gets too small - in that case create a window
+        // queryTimeWindowSizeMs large (the consumer might need to wait a bit
+        // for the window to be ready for reading).
+        Timestamp windowEnd = now.plus(-confidenceWindowSizeMs, ChronoUnit.MILLIS);
+        if (windowEnd.compareTo(windowStart) < 0) {
+            windowEnd = windowStart.plus(queryTimeWindowSizeMs, ChronoUnit.MILLIS);
+        }
+
+        return new TaskState(windowStart, windowEnd, Optional.empty());
     }
 
     /* If the state is before |minimumWindowStart| then this method returns a state
@@ -108,6 +128,12 @@ public TaskState trimTaskState(Timestamp minimumWindowStart, long windowSizeMs)
             return new TaskState(minimumWindowStart, minimumWindowStart.plus(windowSizeMs, ChronoUnit.MILLIS), Optional.empty());
         }
 
-        return this;
+        // Trim the start of the window with minimumWindowStart.
+        Timestamp newWindowStart = windowStart;
+        if (newWindowStart.compareTo(minimumWindowStart) < 0) {
+            newWindowStart = minimumWindowStart;
+        }
+
+        return new TaskState(newWindowStart, windowEnd, lastConsumedChangeId);
     }
 }

scylla-cdc-base/src/main/java/com/scylladb/cdc/model/worker/Worker.java

@@ -48,9 +48,10 @@ private static GenerationId getGenerationIdOfStreams(Map<TaskId, SortedSet<Strea
      * All streams are assumed to belong to the same generation and the initial
      * state is build based on the ID of this generation.
      */
-    private static TaskState getInitialStateForStreams(Map<TaskId, SortedSet<StreamId>> groupedStreams,
-                                                       long windowSizeMs) {
-        return TaskState.createInitialFor(getGenerationIdOfStreams(groupedStreams), windowSizeMs);
+    private TaskState getInitialStateForStreams(Map<TaskId, SortedSet<StreamId>> groupedStreams) {
+        Timestamp now = new Timestamp(Date.from(workerConfiguration.getClock().instant()));
+
+        return TaskState.createInitialFor(getGenerationIdOfStreams(groupedStreams), now, workerConfiguration.confidenceWindowSizeMs, workerConfiguration.queryTimeWindowSizeMs);
     }
 
     /*
@@ -64,7 +65,7 @@ private static TaskState getInitialStateForStreams(Map<TaskId, SortedSet<StreamI
      */
     private Stream<Task> createTasksWithState(Map<TaskId, SortedSet<StreamId>> groupedStreams) throws ExecutionException, InterruptedException {
         Map<TaskId, TaskState> states = workerConfiguration.transport.getTaskStates(groupedStreams.keySet());
-        TaskState initialState = getInitialStateForStreams(groupedStreams, workerConfiguration.queryTimeWindowSizeMs);
+        TaskState initialState = getInitialStateForStreams(groupedStreams);
 
         Set<TableName> tableNames = groupedStreams.keySet().stream().map(TaskId::getTable).collect(Collectors.toSet());
         Date now = Date.from(workerConfiguration.getClock().instant());

scylla-cdc-base/src/test/java/com/scylladb/cdc/cql/MockWorkerCQL.java

@@ -119,6 +119,10 @@ public boolean isReaderFinished(Task task) {
         return finishedReaders.contains(task);
     }
 
+    public Set<Task> getFinishedReaders() {
+        return finishedReaders;
+    }
+
     public void setCQLErrorStrategy(ErrorInject errorStrategy) {
         this.cqlErrorStrategy = Preconditions.checkNotNull(errorStrategy);
     }