Implement set timer by time stamp using Java.time.ZonedDateTime (#106)

* Implement set timer by time stamp

* remove unnecessary check for timer

* add default implementation to avoid breaking change

* update exception message

Author: kaibocai

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

@@ -5,6 +5,7 @@
 import javax.annotation.Nullable;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.ZonedDateTime;
 import java.util.Arrays;
 import java.util.List;
 
@@ -155,6 +156,20 @@ default Task<Task<?>> anyOf(Task<?>... tasks) {
      */
     Task<Void> createTimer(Duration delay);
 
+    /**
+     * Creates a durable timer that expires after the specified timestamp with specific zone.
+     * <p>
+     * Specifying a long delay (for example, a delay of a few days or more) may result in the creation of multiple,
+     * internally-managed durable timers. The orchestration code doesn't need to be aware of this behavior. However,
+     * it may be visible in framework logs and the stored history state.
+     *
+     * @param zonedDateTime timestamp with specific zone when the timer should expire
+     * @return a new {@code Task} that completes after the specified delay
+     */
+    default Task<Void> createTimer(ZonedDateTime zonedDateTime) {
+        throw new UnsupportedOperationException("This method is not implemented. Please use an orchestration context class that has an implementation for this method.");
+    }
+
     /**
      * Transitions the orchestration into the {@link OrchestrationRuntimeStatus#COMPLETED} state with the given output.
      *

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

@@ -10,6 +10,7 @@
 import javax.annotation.Nullable;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.ZonedDateTime;
 import java.util.*;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
@@ -548,6 +549,20 @@ public Task<Void> createTimer(Duration duration) {
 
             int id = this.sequenceNumber++;
             Instant fireAt = this.currentInstant.plus(duration);
+            return createInstantTimer(id, fireAt);
+        }
+
+        @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);
+        }
+
+        private Task<Void> createInstantTimer(int id, Instant fireAt) {
             Timestamp ts = DataConverter.getTimestampFromInstant(fireAt);
             this.pendingActions.put(id, OrchestratorAction.newBuilder()
                     .setId(id)

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

@@ -3,8 +3,7 @@
 package com.microsoft.durabletask;
 
 import java.io.IOException;
-import java.time.Duration;
-import java.time.Instant;
+import java.time.*;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -88,6 +87,30 @@ void singleTimer() throws IOException, TimeoutException {
         }
     }
 
+    @Test
+    void singleTimeStampTimer() throws IOException, TimeoutException {
+        final String orchestratorName = "SingleTimeStampTimer";
+        final Duration delay = Duration.ofSeconds(3);
+        final ZonedDateTime zonedDateTime = ZonedDateTime.of(LocalDateTime.now().plusSeconds(delay.getSeconds()), ZoneId.systemDefault());
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> ctx.createTimer(zonedDateTime).await())
+                .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);
+        }
+    }
+
     @Test
     void isReplaying() throws IOException, InterruptedException, TimeoutException {
         final String orchestratorName = "SingleTimer";