CFP Time based checkpoint strategy

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/ChangeFeedCheckpointStrategyTests.java

@@ -0,0 +1,85 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos;
+
+import com.azure.cosmos.models.ChangeFeedProcessorOptions;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.time.Duration;
+
+public class ChangeFeedCheckpointStrategyTests {
+
+    @Test(groups = {"unit"})
+    public void optionsDefaultCheckpointStrategyIsNull() {
+        ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions();
+
+        Assert.assertNull(options.getCheckpointStrategy());
+    }
+
+    @Test(groups = {"unit"})
+    public void setCheckpointStrategyNullThrows() {
+        ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions();
+
+        Assert.assertThrows(NullPointerException.class, () -> options.setCheckpointStrategy(null));
+    }
+
+    @Test(groups = {"unit"})
+    public void timeIntervalCheckpointStrategyRequiresPositiveDuration() {
+        Assert.assertThrows(NullPointerException.class, () -> new TimeIntervalCheckpointStrategy(null));
+        Assert.assertThrows(IllegalArgumentException.class, () -> new TimeIntervalCheckpointStrategy(Duration.ZERO));
+        Assert.assertThrows(IllegalArgumentException.class, () -> new TimeIntervalCheckpointStrategy(Duration.ofSeconds(-1)));
+    }
+
+    @Test(groups = {"unit"})
+    public void builderRejectsTimeIntervalCheckpointDelayGreaterThanOrEqualToLeaseExpirationInterval() {
+        ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions()
+            .setLeaseExpirationInterval(Duration.ofSeconds(30))
+            .setCheckpointStrategy(new TimeIntervalCheckpointStrategy(Duration.ofSeconds(30)));
+
+        ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorBuilder()
+            .hostName("host-1")
+            .feedContainer(Mockito.mock(CosmosAsyncContainer.class))
+            .leaseContainer(Mockito.mock(CosmosAsyncContainer.class))
+            .handleChanges(changes -> {
+            })
+            .options(options);
+
+        Assert.assertThrows(IllegalArgumentException.class, builder::buildChangeFeedProcessor);
+    }
+
+    @Test(groups = {"unit"})
+    public void builderAcceptsTimeIntervalCheckpointDelayLessThanLeaseExpirationInterval() {
+        ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions()
+            .setLeaseExpirationInterval(Duration.ofSeconds(30))
+            .setCheckpointStrategy(new TimeIntervalCheckpointStrategy(Duration.ofSeconds(10)));
+
+        ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorBuilder()
+            .hostName("host-1")
+            .feedContainer(Mockito.mock(CosmosAsyncContainer.class))
+            .leaseContainer(Mockito.mock(CosmosAsyncContainer.class))
+            .handleChanges(changes -> {
+                if (changes == null) {
+                    throw new IllegalStateException("changes should not be null");
+                }
+            })
+            .options(options);
+
+        try {
+            Method method = ChangeFeedProcessorBuilder.class.getDeclaredMethod("validateChangeFeedProcessorOptions");
+            method.setAccessible(true);
+            method.invoke(builder);
+        } catch (InvocationTargetException e) {
+            Assert.fail("Expected no exception but got: " + e.getCause(), e.getCause());
+        } catch (Exception e) {
+            Assert.fail("Expected no exception but got: " + e.getMessage(), e);
+        }
+    }
+}
+
+
+

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/changefeed/common/AutoCheckpointerTests.java

@@ -0,0 +1,154 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.implementation.changefeed.common;
+
+import com.azure.cosmos.implementation.changefeed.ChangeFeedObserver;
+import com.azure.cosmos.implementation.changefeed.ChangeFeedObserverCloseReason;
+import com.azure.cosmos.implementation.changefeed.ChangeFeedObserverContext;
+import com.azure.cosmos.implementation.changefeed.CheckpointFrequency;
+import com.azure.cosmos.implementation.changefeed.Lease;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+
+public class AutoCheckpointerTests {
+
+    @SuppressWarnings("unchecked")
+    private static <T> T getField(Object target, String fieldName) throws Exception {
+        Field field = target.getClass().getDeclaredField(fieldName);
+        field.setAccessible(true);
+        return (T) field.get(target);
+    }
+
+    private static void setField(Object target, String fieldName, Object value) throws Exception {
+        Field field = target.getClass().getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(target, value);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Mono<Void> invokeCheckpointIfIntervalElapsed(AutoCheckpointer<?> autoCheckpointer) throws Exception {
+        Method method = AutoCheckpointer.class.getDeclaredMethod("checkpointIfIntervalElapsed");
+        method.setAccessible(true);
+        return (Mono<Void>) method.invoke(autoCheckpointer);
+    }
+
+    @Test(groups = {"unit"})
+    @SuppressWarnings("unchecked")
+    public void everyBatchCheckpointStrategyCheckpointsAfterEachBatch() {
+        ChangeFeedObserver<String> observer = Mockito.mock(ChangeFeedObserver.class);
+        ChangeFeedObserverContext<String> context = Mockito.mock(ChangeFeedObserverContext.class);
+
+        Mockito.when(observer.processChanges(any(), any())).thenReturn(Mono.empty());
+        Mockito.when(context.checkpoint()).thenReturn(Mono.empty());
+
+        AutoCheckpointer<String> autoCheckpointer = new AutoCheckpointer<>(new CheckpointFrequency(), observer);
+        autoCheckpointer.open(context);
+
+        autoCheckpointer.processChanges(context, Collections.singletonList("doc")).block();
+
+        Mockito.verify(context, times(1)).checkpoint();
+        autoCheckpointer.close(context, ChangeFeedObserverCloseReason.SHUTDOWN);
+    }
+
+    @Test(groups = {"unit"})
+    @SuppressWarnings("unchecked")
+    public void timeIntervalCheckpointStrategyUsesBackgroundTimer() throws InterruptedException {
+        ChangeFeedObserver<String> observer = Mockito.mock(ChangeFeedObserver.class);
+        ChangeFeedObserverContext<String> context = Mockito.mock(ChangeFeedObserverContext.class);
+
+        Mockito.when(observer.processChanges(any(), any())).thenReturn(Mono.empty());
+        Mockito.when(context.checkpoint()).thenReturn(Mono.empty());
+
+        CheckpointFrequency checkpointFrequency = new CheckpointFrequency().withTimeInterval(Duration.ofMillis(50));
+        AutoCheckpointer<String> autoCheckpointer = new AutoCheckpointer<>(checkpointFrequency, observer);
+        autoCheckpointer.open(context);
+
+        autoCheckpointer.processChanges(context, Collections.singletonList("doc")).block();
+        Thread.sleep(150);
+
+        Mockito.verify(context, Mockito.atLeastOnce()).checkpoint();
+        autoCheckpointer.close(context, ChangeFeedObserverCloseReason.SHUTDOWN);
+    }
+
+    @Test(groups = {"unit"})
+    @SuppressWarnings("unchecked")
+    public void openFailureDoesNotStartIntervalTimer() throws Exception {
+        ChangeFeedObserver<String> observer = Mockito.mock(ChangeFeedObserver.class);
+        ChangeFeedObserverContext<String> context = Mockito.mock(ChangeFeedObserverContext.class);
+
+        Mockito.doThrow(new IllegalStateException("open failed")).when(observer).open(any());
+
+        AutoCheckpointer<String> autoCheckpointer = new AutoCheckpointer<>(
+            new CheckpointFrequency().withTimeInterval(Duration.ofMillis(20)), observer);
+
+        Assert.assertThrows(IllegalStateException.class, () -> autoCheckpointer.open(context));
+
+        Disposable intervalDisposable = getField(autoCheckpointer, "intervalCheckpointDisposable");
+        Assert.assertNull(intervalDisposable);
+    }
+
+    @Test(groups = {"unit"})
+    @SuppressWarnings("unchecked")
+    public void intervalCheckpointFailureDoesNotTerminateIntervalStream() throws InterruptedException {
+        ChangeFeedObserver<String> observer = Mockito.mock(ChangeFeedObserver.class);
+        ChangeFeedObserverContext<String> context = Mockito.mock(ChangeFeedObserverContext.class);
+
+        Mockito.when(observer.processChanges(any(), any())).thenReturn(Mono.empty());
+        Mockito.when(context.checkpoint())
+            .thenReturn(Mono.error(new IllegalStateException("checkpoint failure")), Mono.empty(), Mono.empty());
+
+        AutoCheckpointer<String> autoCheckpointer = new AutoCheckpointer<>(
+            new CheckpointFrequency().withTimeInterval(Duration.ofMillis(40)), observer);
+        autoCheckpointer.open(context);
+
+        autoCheckpointer.processChanges(context, Collections.singletonList("doc")).block();
+        Thread.sleep(240);
+
+        Mockito.verify(context, Mockito.atLeast(2)).checkpoint();
+        autoCheckpointer.close(context, ChangeFeedObserverCloseReason.SHUTDOWN);
+    }
+
+    @Test(groups = {"unit"})
+    @SuppressWarnings("unchecked")
+    public void checkpointSuccessDoesNotClearNewerProgressArrivingDuringInFlightCheckpoint() throws Exception {
+        ChangeFeedObserver<String> observer = Mockito.mock(ChangeFeedObserver.class);
+        ChangeFeedObserverContext<String> context = Mockito.mock(ChangeFeedObserverContext.class);
+
+        Mockito.when(observer.processChanges(any(), any())).thenReturn(Mono.empty());
+
+        Sinks.One<Lease> firstCheckpoint = Sinks.one();
+        Mockito.when(context.checkpoint())
+            .thenReturn(firstCheckpoint.asMono(), Mono.just(Mockito.mock(Lease.class)));
+
+        AutoCheckpointer<String> autoCheckpointer = new AutoCheckpointer<>(
+            new CheckpointFrequency().withTimeInterval(Duration.ofSeconds(1)), observer);
+
+        setField(autoCheckpointer, "lastCheckpointNanoTime", System.nanoTime() - Duration.ofSeconds(2).toNanos());
+
+        autoCheckpointer.processChanges(context, Collections.singletonList("batch-1")).subscribe();
+        autoCheckpointer.processChanges(context, Collections.singletonList("batch-2")).subscribe();
+
+        Mockito.verify(context, times(1)).checkpoint();
+
+        firstCheckpoint.tryEmitValue(Mockito.mock(Lease.class));
+        Thread.sleep(30);
+
+        setField(autoCheckpointer, "lastCheckpointNanoTime", System.nanoTime() - Duration.ofSeconds(2).toNanos());
+        invokeCheckpointIfIntervalElapsed(autoCheckpointer).block();
+
+        Mockito.verify(context, times(2)).checkpoint();
+    }
+}

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/ChangeFeedCheckpointStrategy.java

@@ -0,0 +1,13 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos;
+
+/**
+ * Base type for checkpoint strategies used by {@link ChangeFeedProcessor}.
+ */
+public abstract class ChangeFeedCheckpointStrategy {
+    ChangeFeedCheckpointStrategy() {
+    }
+}
+

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/ChangeFeedProcessorBuilder.java

@@ -403,6 +403,16 @@ private void validateChangeFeedProcessorOptions() {
             //  force a lot of resets and lead to a poor overall performance of ChangeFeedProcessor.
             throw new IllegalArgumentException("changeFeedProcessorOptions: expecting leaseRenewInterval less than leaseExpirationInterval");
         }
+
+        ChangeFeedCheckpointStrategy checkpointStrategy = this.changeFeedProcessorOptions.getCheckpointStrategy();
+        if (checkpointStrategy instanceof TimeIntervalCheckpointStrategy) {
+            if (((TimeIntervalCheckpointStrategy)checkpointStrategy).getMaxCheckpointDelay()
+                .compareTo(this.changeFeedProcessorOptions.getLeaseExpirationInterval()) >= 0) {
+                throw new IllegalArgumentException(
+                    "changeFeedProcessorOptions: expecting checkpointStrategy.maxCheckpointDelay less than leaseExpirationInterval");
+            }
+        }
+
         //  Some extra checks for all versions and deletes mode
         if (ChangeFeedMode.FULL_FIDELITY.equals(changeFeedMode)) {
             if (this.changeFeedProcessorOptions.getStartTime() != null) {

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/EveryBatchCheckpointStrategy.java

@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos;
+
+/**
+ * Checkpoint strategy that writes a checkpoint after each processed batch.
+ */
+public final class EveryBatchCheckpointStrategy extends ChangeFeedCheckpointStrategy {
+    /**
+     * Creates a new every-batch checkpoint strategy.
+     */
+    public EveryBatchCheckpointStrategy() {
+    }
+}
+

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/TimeIntervalCheckpointStrategy.java

@@ -0,0 +1,39 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos;
+
+import java.time.Duration;
+
+import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
+
+/**
+ * Checkpoint strategy that bounds the replay window by a maximum time interval.
+ */
+public final class TimeIntervalCheckpointStrategy extends ChangeFeedCheckpointStrategy {
+    private final Duration maxCheckpointDelay;
+
+    /**
+     * Creates a new time-interval checkpoint strategy.
+     *
+     * @param maxCheckpointDelay the maximum allowed replay window duration.
+     */
+    public TimeIntervalCheckpointStrategy(Duration maxCheckpointDelay) {
+        checkNotNull(maxCheckpointDelay, "Argument 'maxCheckpointDelay' can not be null");
+        if (maxCheckpointDelay.isZero() || maxCheckpointDelay.isNegative()) {
+            throw new IllegalArgumentException("Argument 'maxCheckpointDelay' must be positive");
+        }
+
+        this.maxCheckpointDelay = maxCheckpointDelay;
+    }
+
+    /**
+     * Gets the maximum replay window.
+     *
+     * @return the maximum replay window.
+     */
+    public Duration getMaxCheckpointDelay() {
+        return this.maxCheckpointDelay;
+    }
+}
+