Break long timers into smaller timers (#122)

* Break long timers into smaller timers

* Update CHANGELOG.md

* Break up long timer for zonedDateTime and verify timer count in tests

* Move down new createTimer function

Author: kamperiadis

CHANGELOG.md

@@ -4,7 +4,7 @@
 * Support Suspend and Resume Client APIs ([#104](https://github.com/microsoft/durabletask-java/issues/104))
 * Fix the potential NPE issue of `DurableTaskClient terminate` method ([#104](https://github.com/microsoft/durabletask-java/issues/104))
 * Add waitForCompletionOrCreateCheckStatusResponse client API ([#115](https://github.com/microsoft/durabletask-java/pull/115))
-
+* Support long timers by breaking up into smaller timers ([#114](https://github.com/microsoft/durabletask-java/issues/114))
 
 ## v1.0.0
 

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

@@ -11,6 +11,7 @@
 
 import io.grpc.*;
 
+import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -22,12 +23,14 @@
 public final class DurableTaskGrpcWorker implements AutoCloseable {
     private static final int DEFAULT_PORT = 4001;
     private static final Logger logger = Logger.getLogger(DurableTaskGrpcWorker.class.getPackage().getName());
+    private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3);
 
     private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
     private final HashMap<String, TaskActivityFactory> activityFactories = new HashMap<>();
 
     private final ManagedChannel managedSidecarChannel;
     private final DataConverter dataConverter;
+    private final Duration maximumTimerInterval;
 
     private final TaskHubSidecarServiceBlockingStub sidecarClient;
 
@@ -57,6 +60,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
 
         this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
         this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
+        this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval : DEFAULT_MAXIMUM_TIMER_INTERVAL;
     }
 
     /**
@@ -108,6 +112,7 @@ public void startAndBlock() {
         TaskOrchestrationExecutor taskOrchestrationExecutor = new TaskOrchestrationExecutor(
                 this.orchestrationFactories,
                 this.dataConverter,
+                this.maximumTimerInterval,
                 logger);
         TaskActivityExecutor taskActivityExecutor = new TaskActivityExecutor(
                 this.activityFactories,

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -4,6 +4,7 @@
 
 import io.grpc.Channel;
 
+import java.time.Duration;
 import java.util.HashMap;
 
 /**
@@ -15,6 +16,7 @@ public final class DurableTaskGrpcWorkerBuilder {
     int port;
     Channel channel;
     DataConverter dataConverter;
+    Duration maximumTimerInterval;
 
     /**
      * Adds an orchestration factory to be used by the constructed {@link DurableTaskGrpcWorker}.
@@ -99,6 +101,18 @@ public DurableTaskGrpcWorkerBuilder dataConverter(DataConverter dataConverter) {
         return this;
     }
 
+    /**
+     * Sets the maximum timer interval. If not specified, the default maximum timer interval duration will be used.
+     * The default maximum timer interval duration is 3 days.
+     *
+     * @param maximumTimerInterval the maximum timer interval
+     * @return this builder object
+     */
+    public DurableTaskGrpcWorkerBuilder maximumTimerInterval(Duration maximumTimerInterval) {
+        this.maximumTimerInterval = maximumTimerInterval;
+        return this;
+    }
+
     /**
      * Initializes a new {@link DurableTaskGrpcWorker} object with the settings specified in the current builder object.
      * @return a new {@link DurableTaskGrpcWorker} object

client/src/main/java/com/microsoft/durabletask/OrchestrationRunner.java

@@ -6,6 +6,7 @@
 import com.google.protobuf.StringValue;
 import com.microsoft.durabletask.implementation.protobuf.OrchestratorService;
 
+import java.time.Duration;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.logging.Logger;
@@ -18,6 +19,7 @@
  */
 public final class OrchestrationRunner {
     private static final Logger logger = Logger.getLogger(OrchestrationRunner.class.getPackage().getName());
+    private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3);
 
     private OrchestrationRunner() {
     }
@@ -126,6 +128,7 @@ public TaskOrchestration create() {
         TaskOrchestrationExecutor taskOrchestrationExecutor = new TaskOrchestrationExecutor(
                 orchestrationFactories,
                 new JacksonDataConverter(),
+                DEFAULT_MAXIMUM_TIMER_INTERVAL,
                 logger);
 
         // TODO: Error handling

client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java

@@ -24,13 +24,16 @@ final class TaskOrchestrationExecutor {
     private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories;
     private final DataConverter dataConverter;
     private final Logger logger;
+    private final Duration maximumTimerInterval;
 
     public TaskOrchestrationExecutor(
             HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
             DataConverter dataConverter,
+            Duration maximumTimerInterval,
             Logger logger) {
         this.orchestrationFactories = orchestrationFactories;
         this.dataConverter = dataConverter;
+        this.maximumTimerInterval = maximumTimerInterval;
         this.logger = logger;
     }
 
@@ -77,6 +80,7 @@ private class ContextImplTask implements TaskOrchestrationContext {
         private final LinkedList<HistoryEvent> unprocessedEvents = new LinkedList<>();
         private final Queue<HistoryEvent> eventsWhileSuspended = new ArrayDeque<>();
         private final DataConverter dataConverter = TaskOrchestrationExecutor.this.dataConverter;
+        private final Duration maximumTimerInterval = TaskOrchestrationExecutor.this.maximumTimerInterval;
         private final Logger logger = TaskOrchestrationExecutor.this.logger;
         private final OrchestrationHistoryIterator historyEventPlayer;
         private int sequenceNumber;
@@ -547,19 +551,27 @@ public Task<Void> createTimer(Duration duration) {
             Helpers.throwIfOrchestratorComplete(this.isComplete);
             Helpers.throwIfArgumentNull(duration, "duration");
 
-            int id = this.sequenceNumber++;
-            Instant fireAt = this.currentInstant.plus(duration);
-            return createInstantTimer(id, fireAt);
+            Instant finalFireAt = this.currentInstant.plus(duration);
+            return createTimer(finalFireAt);
         }
 
         @Override
         public Task<Void> createTimer(ZonedDateTime zonedDateTime) {
             Helpers.throwIfOrchestratorComplete(this.isComplete);
             Helpers.throwIfArgumentNull(zonedDateTime, "zonedDateTime");
 
-            int id = this.sequenceNumber++;
-            Instant fireAt = zonedDateTime.toInstant();
-            return createInstantTimer(id, fireAt);
+            Instant finalFireAt = zonedDateTime.toInstant();
+            return createTimer(finalFireAt);
+        }
+
+        private Task<Void> createTimer(Instant finalFireAt) {
+            Duration remainingTime = Duration.between(this.currentInstant, finalFireAt);
+            while (remainingTime.compareTo(this.maximumTimerInterval) > 0) {
+                Instant nextFireAt = this.currentInstant.plus(this.maximumTimerInterval);
+                createInstantTimer(this.sequenceNumber++, nextFireAt).await();
+                remainingTime = Duration.between(this.currentInstant, finalFireAt);
+            }
+            return createInstantTimer(this.sequenceNumber++, finalFireAt);
         }
 
         private Task<Void> createInstantTimer(int id, Instant fireAt) {

client/src/test/java/com/microsoft/durabletask/IntegrationTestBase.java

@@ -38,6 +38,11 @@ public DurableTaskGrpcWorker buildAndStart() {
             return server;
         }
 
+        public TestDurableTaskWorkerBuilder setMaximumTimerInterval(Duration maximumTimerInterval) {
+            this.innerBuilder.maximumTimerInterval(maximumTimerInterval);
+            return this;
+        }
+
         public TestDurableTaskWorkerBuilder addOrchestrator(
                 String name,
                 TaskOrchestration implementation) {

client/src/test/java/com/microsoft/durabletask/IntegrationTests.java

@@ -7,6 +7,7 @@
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -55,7 +56,7 @@ void emptyOrchestration() throws TimeoutException {
                 instanceId,
                 defaultTimeout,
                 true);
-            
+
             assertNotNull(instance);
             assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
             assertEquals(input, instance.readInputAs(String.class));
@@ -86,6 +87,72 @@ void singleTimer() throws IOException, TimeoutException {
         }
     }
 
+    @Test
+    void longTimer() throws TimeoutException {
+        final String orchestratorName = "LongTimer";
+        final Duration delay = Duration.ofSeconds(7);
+        AtomicInteger counter = new AtomicInteger();
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    counter.incrementAndGet();
+                    ctx.createTimer(delay).await();
+                })
+                .setMaximumTimerInterval(Duration.ofSeconds(3))
+                .buildAndStart();
+
+        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+            Duration timeout = delay.plus(defaultTimeout);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, timeout, false);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+
+            // Verify that the delay actually happened
+            long expectedCompletionSecond = instance.getCreatedAt().plus(delay).getEpochSecond();
+            long actualCompletionSecond = instance.getLastUpdatedAt().getEpochSecond();
+            assertTrue(expectedCompletionSecond <= actualCompletionSecond);
+
+            // Verify that the correct number of timers were created
+            // This should yield 4 (first invocation + replay invocations for internal timers 3s + 3s + 1s)
+            assertEquals(4, counter.get());
+        }
+    }
+
+    @Test
+    void longTimeStampTimer() throws TimeoutException {
+        final String orchestratorName = "LongTimeStampTimer";
+        final Duration delay = Duration.ofSeconds(7);
+        final ZonedDateTime zonedDateTime = ZonedDateTime.of(LocalDateTime.now().plusSeconds(delay.getSeconds()), ZoneId.systemDefault());
+
+        AtomicInteger counter = new AtomicInteger();
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    counter.incrementAndGet();
+                    ctx.createTimer(zonedDateTime).await();
+                })
+                .setMaximumTimerInterval(Duration.ofSeconds(3))
+                .buildAndStart();
+
+        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+            Duration timeout = delay.plus(defaultTimeout);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, timeout, false);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+
+            // Verify that the delay actually happened
+            long expectedCompletionSecond = zonedDateTime.toInstant().getEpochSecond();
+            long actualCompletionSecond = instance.getLastUpdatedAt().getEpochSecond();
+            assertTrue(expectedCompletionSecond <= actualCompletionSecond);
+
+            // Verify that the correct number of timers were created
+            // This should yield 4 (first invocation + replay invocations for internal timers 3s + 3s + 1s)
+            assertEquals(4, counter.get());
+        }
+    }
+
     @Test
     void singleTimeStampTimer() throws IOException, TimeoutException {
         final String orchestratorName = "SingleTimeStampTimer";