Adds additional rollup delay tests

Signed-off-by: Clay Downs <downsrob@amazon.com>

Author: downsrob

src/main/kotlin/org/opensearch/indexmanagement/rollup/model/Rollup.kt

@@ -348,11 +348,7 @@ data class Rollup(
             // TODO: Make startTime public in Job Scheduler so we can just directly check the value
             if (seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || primaryTerm == SequenceNumbers.UNASSIGNED_PRIMARY_TERM) {
                 if (schedule is IntervalSchedule) {
-                    schedule = if (schedule.delay == null) {
-                        IntervalSchedule(Instant.now(), schedule.interval, schedule.unit)
-                    } else {
-                        IntervalSchedule(Instant.now(), schedule.interval, schedule.unit, schedule.delay)
-                    }
+                    schedule = IntervalSchedule(Instant.now(), schedule.interval, schedule.unit, schedule.delay ?: 0)
                 }
             }
             return Rollup(

src/test/kotlin/org/opensearch/indexmanagement/rollup/model/RollupTests.kt

@@ -106,7 +106,7 @@ class RollupTests : OpenSearchTestCase() {
         randomRollup().copy(delay = null)
     }
 
-    fun `test rollup applies to continuous rollups only`() {
+    fun `test delay applies to continuous rollups only`() {
         // Continuous rollup schedule matches delay
         val newDelay: Long = 500
         val continuousRollup = randomRollup().copy(

src/test/kotlin/org/opensearch/indexmanagement/rollup/model/WriteableTests.kt

@@ -128,7 +128,7 @@ class WriteableTests : OpenSearchTestCase() {
     }
 
     fun `test rollup as stream`() {
-        val rollup = randomRollup()
+        val rollup = randomRollup().copy(delay = randomLongBetween(0, 60000000))
         val out = BytesStreamOutput().also { rollup.writeTo(it) }
         val sin = StreamInput.wrap(out.bytes().toBytesRef().bytes)
         val streamedRollup = Rollup(sin)

src/test/kotlin/org/opensearch/indexmanagement/rollup/model/XContentTests.kt

@@ -126,7 +126,7 @@ class XContentTests : OpenSearchTestCase() {
     }
 
     fun `test rollup parsing with type`() {
-        val rollup = randomRollup()
+        val rollup = randomRollup().copy(delay = randomLongBetween(0, 60000000))
         val rollupString = rollup.toJsonString()
         val parser = parserWithType(rollupString)
         val parsedRollup = parser.parseWithType(rollup.id, rollup.seqNo, rollup.primaryTerm, Rollup.Companion::parse)
@@ -135,7 +135,7 @@ class XContentTests : OpenSearchTestCase() {
     }
 
     fun `test rollup parsing without type`() {
-        val rollup = randomRollup()
+        val rollup = randomRollup().copy(delay = randomLongBetween(0, 60000000))
         val rollupString = rollup.toJsonString(XCONTENT_WITHOUT_TYPE)
         val parsedRollup = Rollup.parse(parser(rollupString), rollup.id, rollup.seqNo, rollup.primaryTerm)
         // roles are deprecated and not populated in toXContent and parsed as part of parse

src/test/kotlin/org/opensearch/indexmanagement/rollup/runner/RollupRunnerIT.kt

@@ -577,15 +577,15 @@ class RollupRunnerIT : RollupRestTestCase() {
         // Define rollup
         var rollup = randomRollup().copy(
             enabled = true,
-            jobSchedule = IntervalSchedule(Instant.now(), 1, ChronoUnit.MINUTES, 0),
+            jobSchedule = IntervalSchedule(Instant.now(), 1, ChronoUnit.MINUTES),
             jobEnabledTime = Instant.now(),
             sourceIndex = indexName,
             metadataID = null,
             continuous = true,
             delay = delay,
             dimensions = listOf(
                 randomCalendarDateHistogram().copy(
-                    calendarInterval = "1s"
+                    calendarInterval = "5s"
                 )
             )
         )
@@ -598,7 +598,7 @@ class RollupRunnerIT : RollupRestTestCase() {
         // Create rollup job
         rollup = createRollup(rollup = rollup, rollupId = rollup.id)
 
-        val nextExecutionTime = rollup.schedule.getNextExecutionTime(null).toEpochMilli()
+        var nextExecutionTime = rollup.schedule.getNextExecutionTime(null).toEpochMilli()
         val expectedExecutionTime = rollup.jobEnabledTime!!.plusMillis(delay).toEpochMilli()
         val delayIsCorrect = ((expectedExecutionTime - nextExecutionTime) > -500) && ((expectedExecutionTime - nextExecutionTime) < 500)
         assertTrue("Delay was not correctly applied", delayIsCorrect)
@@ -609,20 +609,29 @@ class RollupRunnerIT : RollupRestTestCase() {
             // Still should not have run at this point
             assertFalse("Target rollup index was created before the delay should allow", indexExists(rollup.targetIndex))
         }
-
-        waitFor {
+        val rollupMetadata = waitFor {
             assertTrue("Target rollup index was not created", indexExists(rollup.targetIndex))
             val rollupJob = getRollup(rollupId = rollup.id)
             assertNotNull("Rollup job doesn't have metadata set", rollupJob.metadataID)
             val rollupMetadata = getRollupMetadata(rollupJob.metadataID!!)
             assertNotNull("Rollup metadata not found", rollupMetadata)
-            val nextWindowStartTime: Instant = rollupMetadata.continuous!!.nextWindowStartTime
-            assertTrue("Rollup window not excluding the delay", nextWindowStartTime < Instant.now().minusMillis(delay))
-            assertTrue("Rollup window not updating", nextWindowStartTime > Instant.now().minusMillis(delay + 2000))
+            rollupMetadata
         }
+        nextExecutionTime = rollup.schedule.getNextExecutionTime(null).toEpochMilli()
+        val nextExecutionOffset = (nextExecutionTime - Instant.now().toEpochMilli()) - 60000
+        val nextExecutionIsCorrect = nextExecutionOffset < 5000 && nextExecutionOffset > -5000
+        assertTrue("Next execution time not updated correctly", nextExecutionIsCorrect)
+        val nextWindowStartTime: Instant = rollupMetadata.continuous!!.nextWindowStartTime
+        val nextWindowEndTime: Instant = rollupMetadata.continuous!!.nextWindowEndTime
+        // Assert that after the window was updated, it falls approximately around 'now'
+        assertTrue("Rollup window start time is incorrect", nextWindowStartTime.plusMillis(delay).minusMillis(1000) < Instant.now())
+        assertTrue("Rollup window end time is incorrect", nextWindowEndTime.plusMillis(delay).plusMillis(1000) > Instant.now())
+
+        // window length should be 5 seconds
+        val expectedWindowEnd = nextWindowStartTime.plusMillis(5000)
+        assertEquals("Rollup window length applied incorrectly", expectedWindowEnd, nextWindowEndTime)
     }
 
-    // Tests that the non continuous delay does nothing
     fun `test non continuous delay does nothing`() {
         generateNYCTaxiData("source_runner_ninth")