adding test for loop createTimer

Signed-off-by: salaboy <[email protected]>

Author: salaboy

client/build.gradle

@@ -11,7 +11,7 @@ plugins {
 }
 
 group 'io.dapr'
-version = '1.5.9'
+version = '1.5.10-SNAPSHOT'
 archivesBaseName = 'durabletask-client'
 
 def grpcVersion = '1.69.0'

client/src/main/java/io/dapr/durabletask/DurableTaskClient.java

@@ -4,6 +4,7 @@
 
 import javax.annotation.Nullable;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.concurrent.TimeoutException;
 
 /**
@@ -62,6 +63,7 @@ public String scheduleNewOrchestrationInstance(String orchestratorName, Object i
      */
     public String scheduleNewOrchestrationInstance(String orchestratorName, Object input, String instanceId) {
         NewOrchestrationInstanceOptions options = new NewOrchestrationInstanceOptions()
+                .setStartTime(Instant.now())
                 .setInput(input)
                 .setInstanceId(instanceId);
         return this.scheduleNewOrchestrationInstance(orchestratorName, options);

client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java

@@ -14,9 +14,11 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.ZonedDateTime;
+import java.time.temporal.TemporalField;
 import java.util.*;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -88,10 +90,10 @@ private class ContextImplTask implements TaskOrchestrationContext {
         private String appId;
 
         // LinkedHashMap to maintain insertion order when returning the list of pending actions
-        private final LinkedHashMap<Integer, OrchestratorAction> pendingActions = new LinkedHashMap<>();
-        private final HashMap<Integer, TaskRecord<?>> openTasks = new HashMap<>();
-        private final LinkedHashMap<String, Queue<TaskRecord<?>>> outstandingEvents = new LinkedHashMap<>();
-        private final LinkedList<HistoryEvent> unprocessedEvents = new LinkedList<>();
+        private final Map<Integer, OrchestratorAction> pendingActions = new LinkedHashMap<>();
+        private final Map<Integer, TaskRecord<?>> openTasks = new ConcurrentHashMap<>();
+        private final Map<String, Queue<TaskRecord<?>>> outstandingEvents = new LinkedHashMap<>();
+        private final List<HistoryEvent> unprocessedEvents = Collections.synchronizedList(new LinkedList<>());
         private final Queue<HistoryEvent> eventsWhileSuspended = new ArrayDeque<>();
         private final DataConverter dataConverter = TaskOrchestrationExecutor.this.dataConverter;
         private final Duration maximumTimerInterval = TaskOrchestrationExecutor.this.maximumTimerInterval;
@@ -303,12 +305,12 @@ public <V> Task<V> callActivity(
             }
             TaskFactory<V> taskFactory = () -> {
                 int id = this.sequenceNumber++;
-                
+
                 ScheduleTaskAction scheduleTaskAction = scheduleTaskBuilder.build();
                 OrchestratorAction.Builder actionBuilder = OrchestratorAction.newBuilder()
                         .setId(id)
                         .setScheduleTask(scheduleTaskBuilder);
-                
+
                 if (options != null && options.hasAppID()) {
                     String targetAppId = options.getAppID();
                     TaskRouter actionRouter = TaskRouter.newBuilder()
@@ -317,7 +319,7 @@ public <V> Task<V> callActivity(
                         .build();
                     actionBuilder.setRouter(actionRouter);
                 }
-                
+
                 this.pendingActions.put(id, actionBuilder.build());
 
                 if (!this.isReplaying) {
@@ -410,7 +412,7 @@ public <V> Task<V> callSubOrchestrator(
             if (input instanceof TaskOptions) {
                 throw new IllegalArgumentException("TaskOptions cannot be used as an input. Did you call the wrong method overload?");
             }
-            
+
             String serializedInput = this.dataConverter.serialize(input);
             CreateSubOrchestrationAction.Builder createSubOrchestrationActionBuilder = CreateSubOrchestrationAction.newBuilder().setName(name);
             if (serializedInput != null) {
@@ -466,7 +468,7 @@ public <V> Task<V> waitForExternalEvent(String name, Duration timeout, Class<V>
             int id = this.sequenceNumber++;
 
             CompletableTask<V> eventTask = new ExternalEventTask<>(name, id, timeout);
-            
+
             // Check for a previously received event with the same name
             for (HistoryEvent e : this.unprocessedEvents) {
                 EventRaisedEvent existing = e.getEventRaised();
@@ -648,8 +650,8 @@ public Task<Void> createTimer(ZonedDateTime zonedDateTime) {
         }
 
         private Task<Void> createTimer(Instant finalFireAt) {
-            TimerTask timer = new TimerTask(finalFireAt);
-            return timer;
+            logger.info(">>>> Creating a Timer Task to fire at: "+ finalFireAt);
+            return new TimerTask(finalFireAt);
         }
 
         private CompletableTask<Void> createInstantTimer(int id, Instant fireAt) {
@@ -660,9 +662,11 @@ private CompletableTask<Void> createInstantTimer(int id, Instant fireAt) {
                     .build());
 
             if (!this.isReplaying) {
-                // TODO: Log timer creation, including the expected fire-time
+                logger.info("Creating Instant Timer with id: " + id + " fireAt: " + fireAt);
             }
 
+            logger.info("REPLAY: Creating Instant Timer with id: " + id + " fireAt: " + fireAt);
+
             CompletableTask<Void> timerTask = new CompletableTask<>();
             TaskRecord<Void> record = new TaskRecord<>(timerTask, "(timer)", Void.class);
             this.openTasks.put(id, record);
@@ -702,7 +706,9 @@ public void handleTimerFired(HistoryEvent e) {
 
             if (!this.isReplaying) {
                 // TODO: Log timer fired, including the scheduled fire-time
+                this.logger.info("Firing timer by completing task: "+timerEventId+" expected fire at time: "+ Instant.ofEpochSecond(timerFiredEvent.getFireAt().getSeconds(), timerFiredEvent.getFireAt().getNanos()));
             }
+            this.logger.info("REPLAY: Firing timer by completing task: "+timerEventId+" expected fire at time: "+ Instant.ofEpochSecond(timerFiredEvent.getFireAt().getSeconds(), timerFiredEvent.getFireAt().getNanos()));
 
             CompletableTask<?> task = record.getTask();
             task.complete(null);
@@ -851,7 +857,7 @@ private void addCarryoverEvents(CompleteOrchestrationAction.Builder builder) {
 
             externalEvents.forEach(builder::addCarryoverEvents);
         }
-        
+
         private boolean waitingForEvents() {
             return this.outstandingEvents.size() > 0;
         }
@@ -894,7 +900,7 @@ private void processEvent(HistoryEvent e) {
                         if (factory == null) {
                             throw new IllegalStateException("No factory found for orchestrator: " + executionStarted.getName());
                         }
-                        
+
                         TaskOrchestration orchestrator = factory.create();
                         orchestrator.run(this);
                         break;
@@ -1025,6 +1031,7 @@ private class TimerTask extends CompletableTask<Void> {
 
             public TimerTask(Instant finalFireAt) {
                 super();
+                logger.info("Creating a Timer task at: " + finalFireAt + " -> with current time " + currentInstant );
                 CompletableTask<Void> firstTimer = createTimerTask(finalFireAt);
                 CompletableFuture<Void> timerChain = createTimerChain(finalFireAt, firstTimer.future);
                 this.task = new CompletableTask<>(timerChain);
@@ -1062,7 +1069,10 @@ private CompletableTask<Void> createTimerTask(Instant finalFireAt) {
 
             private void handleSubTimerSuccess() {
                 // check if it is the last timer
+                logger.info(">>>> Comparing current instant "+ currentInstant+" with finalFireAt: " + finalFireAt);
+                logger.info(">>>>Comparison: "+currentInstant.compareTo(finalFireAt));
                 if (currentInstant.compareTo(finalFireAt) >= 0) {
+                    logger.info(">>>> Comparison -> Complete");
                     this.complete(null);
                 }
             }

client/src/test/java/io/dapr/durabletask/IntegrationTests.java

@@ -124,6 +124,53 @@ void singleTimer() throws IOException, TimeoutException {
         }
     }
 
+    @Test
+    void loopWithTimer() throws IOException, TimeoutException {
+        final String orchestratorName = "LoopWithTimer";
+        final Duration delay = Duration.ofSeconds(2);
+        AtomicReferenceArray<LocalDateTime> timestamps = new AtomicReferenceArray<>(100);
+        AtomicInteger counter = new AtomicInteger();
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    for(int i = 0 ; i < 10; i++) {
+                        timestamps.set(counter.get(), LocalDateTime.now());
+                        counter.incrementAndGet();
+                        ctx.createTimer(delay).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 = instance.getCreatedAt().plus(delay).getEpochSecond();
+            long actualCompletionSecond = instance.getLastUpdatedAt().getEpochSecond();
+            assertTrue(expectedCompletionSecond <= actualCompletionSecond);
+
+            // Verify that the correct number of timers were created
+            // ??? Not sure why 65, this seems consistent with what we see in Catalyst
+            assertEquals(65, counter.get());
+
+            // Verify that each timer is the expected length
+            int[] secondsElapsed = new int[timestamps.length()];
+            for (int i = 0; i < timestamps.length() - 1; i++) {
+                if(timestamps.get(i + 1) != null && timestamps.get(i) != null ) {
+                    secondsElapsed[i] = timestamps.get(i + 1).getSecond() - timestamps.get(i).getSecond();
+                }else{
+                    secondsElapsed[i] = -1;
+                }
+            }
+            assertEquals(secondsElapsed[0], 2);
+
+        }
+    }
+
     @Test
     void longTimer() throws TimeoutException {
         final String orchestratorName = "LongTimer";