branch-3.0: [Fix](Job)Fix the Calculation of the First Trigger Time a…

…nd Add a Single-Time Scheduling Compensation Logic #44268 (#44402) Cherry-picked from #44268 Co-authored-by: Calvin Kirs <[email protected]>
apache · Nov 29, 2024 · 9ff34b2 · 9ff34b2
1 parent c67a806
commit 9ff34b2
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/base/JobExecutionConfiguration.java b/fe/fe-core/src/main/java/org/apache/doris/job/base/JobExecutionConfiguration.java
@@ -172,9 +172,10 @@ private List<Long> getExecutionDelaySeconds(long windowStartTimeMs, long windowE
         long firstTriggerTime = windowStartTimeMs + (intervalMs - ((windowStartTimeMs - startTimeMs)
                 % intervalMs)) % intervalMs;
         if (firstTriggerTime < currentTimeMs) {
-            firstTriggerTime += intervalMs;
+            // Calculate how many intervals to add to get the largest trigger time < currentTimeMs
+            long intervalsToAdd = (currentTimeMs - firstTriggerTime) / intervalMs;
+            firstTriggerTime += intervalsToAdd * intervalMs;
         }
-
         if (firstTriggerTime > windowEndTimeMs) {
             return timestamps; // Return an empty list if there won't be any trigger time
         }

diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/base/JobExecutionConfigurationTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/base/JobExecutionConfigurationTest.java
@@ -60,14 +60,22 @@ public void testGetTriggerDelayTimesRecurring() {
         Assertions.assertArrayEquals(new Long[]{100L, 700L}, delayTimes.toArray());
         delayTimes = configuration.getTriggerDelayTimes(
                 200000L, 0L, 1100000L);
-        Assertions.assertEquals(1, delayTimes.size());
-        Assertions.assertArrayEquals(new Long[]{500L}, delayTimes.toArray());
+        Assertions.assertEquals(2, delayTimes.size());
+        Assertions.assertArrayEquals(new Long[]{0L, 500L}, delayTimes.toArray());
         delayTimes = configuration.getTriggerDelayTimes(
                 1001000L, 0L, 1000000L);
         Assertions.assertEquals(1, delayTimes.size());
         timerDefinition.setStartTimeMs(2000L);
         timerDefinition.setIntervalUnit(IntervalUnit.SECOND);
         Assertions.assertArrayEquals(new Long[]{2L, 12L}, configuration.getTriggerDelayTimes(100000L, 100000L, 120000L).toArray());
+
+        timerDefinition.setIntervalUnit(IntervalUnit.SECOND);
+        long second = 1000L;
+        timerDefinition.setStartTimeMs(second);
+        timerDefinition.setInterval(1L);
+        Assertions.assertEquals(3, configuration.getTriggerDelayTimes(second * 5 + 10L, second * 3, second * 7).size());
+        Assertions.assertEquals(3, configuration.getTriggerDelayTimes(second * 5, second * 5, second * 7).size());
+
     }
 
     @Test