Support defining IntervalSchedule in seconds

sample-extension-plugin/build.gradle

@@ -37,6 +37,9 @@ repositories {
 
 dependencies {
     compileOnly project(path: ":${rootProject.name}-spi", configuration: 'shadow')
+    testImplementation('org.awaitility:awaitility:4.3.0') {
+        exclude(group: 'org.hamcrest', module: 'hamcrest')
+    }
 }
 
 def es_tmp_dir = rootProject.file('build/private/es_tmp').absoluteFile
@@ -112,6 +115,7 @@ testClusters.integTest {
         }
     }
     setting 'path.repo', repo.absolutePath
+    setting 'plugins.jobscheduler.sweeper.period', '1s'
 }
 
 String baseName = "jobSchedulerBwcCluster"
@@ -149,6 +153,7 @@ String bwcDownloadUrl = "https://aws.oss.sonatype.org/service/local/artifact/mav
             }))
             setting 'path.repo', "${buildDir}/cluster/shared/repo/${baseName}"
             setting 'http.content_type.required', 'true'
+            setting 'plugins.jobscheduler.sweeper.period', '1s'
         }
     }
 }

sample-extension-plugin/src/main/java/org/opensearch/jobscheduler/sampleextension/SampleExtensionRestHandler.java

@@ -90,7 +90,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
 
             Schedule schedule;
             if (interval != null) {
-                schedule = new IntervalSchedule(Instant.now(), Integer.parseInt(interval), ChronoUnit.MINUTES);
+                schedule = new IntervalSchedule(Instant.now(), Integer.parseInt(interval), ChronoUnit.SECONDS);
             } else {
                 schedule = new CronSchedule(cron, ZoneId.systemDefault());
             }

sample-extension-plugin/src/main/java/org/opensearch/jobscheduler/sampleextension/SampleJobRunner.java

@@ -143,7 +143,7 @@ private void runTaskForIntegrationTests(SampleJobParameter jobParameter) {
 
     private void runTaskForLockIntegrationTests(SampleJobParameter jobParameter) throws InterruptedException {
         if (jobParameter.getName().equals("sample-job-lock-test-it")) {
-            Thread.sleep(180000);
+            Thread.sleep(10000);
         }
     }
 }

sample-extension-plugin/src/test/java/org/opensearch/jobscheduler/sampleextension/SampleExtensionIntegTestCase.java

@@ -37,6 +37,7 @@
 import org.opensearch.common.xcontent.LoggingDeprecationHandler;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.jobscheduler.spi.LockModel;
 import org.opensearch.jobscheduler.spi.schedule.CronSchedule;
 import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule;
 import org.opensearch.core.rest.RestStatus;
@@ -59,8 +60,11 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.Timer;
-import java.util.TimerTask;
+import static org.awaitility.Awaitility.await;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class SampleExtensionIntegTestCase extends OpenSearchRestTestCase {
 
@@ -179,10 +183,6 @@ protected SampleJobParameter createWatcherJob(String jobId, SampleJobParameter j
         return createWatcherJobWithClient(client(), jobId, jobParameter);
     }
 
-    protected String createWatcherJobJson(String jobId, String jobParameter) throws IOException {
-        return createWatcherJobJsonWithClient(client(), jobId, jobParameter);
-    }
-
     protected SampleJobParameter createWatcherJobWithClient(RestClient client, String jobId, SampleJobParameter jobParameter)
         throws IOException {
         Map<String, String> params = getJobParameterAsMap(jobId, jobParameter);
@@ -315,7 +315,7 @@ protected SampleJobParameter getJobParameter(RestClient client, String jobId) th
                 new IntervalSchedule(
                     Instant.ofEpochMilli(Long.parseLong(((Map<String, Object>) jobSchedule.get("interval")).get("start_time").toString())),
                     Integer.parseInt(((Map<String, Object>) jobSchedule.get("interval")).get("period").toString()),
-                    ChronoUnit.MINUTES
+                    ChronoUnit.SECONDS
                 )
 
             );
@@ -338,7 +338,7 @@ protected void deleteTestIndex(String index) throws IOException {
         deleteIndex(index);
     }
 
-    protected long countRecordsInTestIndex(String index) throws IOException {
+    protected int countRecordsInTestIndex(String index) throws IOException {
         String entity = """
             {
                 "query": {
@@ -362,69 +362,29 @@ protected long countRecordsInTestIndex(String index) throws IOException {
         return Integer.parseInt(responseJson.get("count").toString());
     }
 
-    protected void waitAndCreateWatcherJob(String prevIndex, String jobId, SampleJobParameter jobParameter) {
-        Timer timer = new Timer();
-        TimerTask timerTask = new TimerTask() {
-            private int timeoutCounter = 0;
-
-            @Override
-            public void run() {
-                try {
-                    long count = countRecordsInTestIndex(prevIndex);
-                    ++timeoutCounter;
-                    if (count == 1) {
-                        createWatcherJob(jobId, jobParameter);
-                        timer.cancel();
-                        timer.purge();
-                    }
-                    if (timeoutCounter >= 24) {
-                        timer.cancel();
-                        timer.purge();
-                    }
-                } catch (IOException ex) {
-                    // do nothing
-                    // suppress exception
-                }
-            }
-        };
-        timer.scheduleAtFixedRate(timerTask, 2000, 5000);
+    protected void waitUntilLockIsAcquiredAndReleased(String jobId) {
+        waitUntilLockIsAcquiredAndReleased(jobId, 20);
     }
 
-    protected void waitAndDeleteWatcherJob(String prevIndex, String jobId) {
-        Timer timer = new Timer();
-        TimerTask timerTask = new TimerTask() {
-            private int timeoutCounter = 0;
-
-            @Override
-            public void run() {
-                try {
-                    long count = countRecordsInTestIndex(prevIndex);
-                    ++timeoutCounter;
-                    if (count == 1) {
-                        deleteWatcherJob(jobId);
-                        timer.cancel();
-                        timer.purge();
-                    }
-                    if (timeoutCounter >= 24) {
-                        timer.cancel();
-                        timer.purge();
-                    }
-                } catch (IOException ex) {
-                    // do nothing
-                    // suppress exception
-                }
+    protected void waitUntilLockIsAcquiredAndReleased(String jobId, int maxTimeInSec) {
+        AtomicLong prevLockAcquiredTime = new AtomicLong(0L);
+        AtomicReference<LockModel> lock = new AtomicReference<>();
+        try {
+            lock.set(getLockByJobId(jobId));
+            if (lock.get() != null && prevLockAcquiredTime.get() == 0L && lock.get().isReleased()) {
+                prevLockAcquiredTime.set(lock.get().getLockTime().toEpochMilli());
             }
-        };
-        timer.scheduleAtFixedRate(timerTask, 2000, 5000);
-    }
-
-    protected long waitAndCountRecords(String index, long waitForInMs) throws Exception {
-        Thread.sleep(waitForInMs);
-        return countRecordsInTestIndex(index);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        await().atMost(maxTimeInSec, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).ignoreExceptions().until(() -> {
+            lock.set(getLockByJobId(jobId));
+            return lock.get() != null && lock.get().getLockTime().toEpochMilli() != prevLockAcquiredTime.get() && lock.get().isReleased();
+        });
     }
 
     @SuppressWarnings("unchecked")
-    protected long getLockTimeByJobId(String jobId) throws IOException {
+    protected LockModel getLockByJobId(String jobId) throws IOException {
         String entity = """
             {
                 "query": {
@@ -439,49 +399,29 @@ protected long getLockTimeByJobId(String jobId) throws IOException {
         Response response = makeRequest(
             client(),
             "POST",
-            "/" + ".opendistro-job-scheduler-lock" + "/_search",
-            Collections.emptyMap(),
+            "/.opendistro-job-scheduler-lock/_search",
+            Map.of("ignore", "404"),
             new StringEntity(entity, ContentType.APPLICATION_JSON)
         );
+        if (response.getStatusLine().getStatusCode() == 404) {
+            return null;
+        }
         Map<String, Object> responseJson = JsonXContent.jsonXContent.createParser(
             NamedXContentRegistry.EMPTY,
             LoggingDeprecationHandler.INSTANCE,
             response.getEntity().getContent()
         ).map();
         List<Map<String, Object>> hits = (List<Map<String, Object>>) ((Map<String, Object>) responseJson.get("hits")).get("hits");
-        if (hits.size() == 0) {
-            return 0L;
+        if (hits.isEmpty()) {
+            return null;
         }
         Map<String, Object> lockSource = (Map<String, Object>) hits.get(0).get("_source");
-        return Long.parseLong(lockSource.get("lock_time").toString());
-    }
-
-    @SuppressWarnings("unchecked")
-    protected boolean doesLockExistByLockTime(long lockTime) throws IOException {
-        String entity = """
-            {
-                "query": {
-                    "match": {
-                        "lock_time": {
-                            "query": %d
-                        }
-                    }
-                }
-            }
-            """.formatted(lockTime);
-        Response response = makeRequest(
-            client(),
-            "POST",
-            "/" + ".opendistro-job-scheduler-lock" + "/_search",
-            Collections.emptyMap(),
-            new StringEntity(entity, ContentType.APPLICATION_JSON)
+        return new LockModel(
+            lockSource.get(LockModel.JOB_INDEX_NAME).toString(),
+            lockSource.get(LockModel.JOB_ID).toString(),
+            Instant.ofEpochMilli(Long.parseLong(lockSource.get(LockModel.LOCK_TIME).toString())),
+            Long.parseLong(lockSource.get(LockModel.LOCK_DURATION).toString()),
+            Boolean.parseBoolean(lockSource.get(LockModel.RELEASED).toString())
         );
-        Map<String, Object> responseJson = JsonXContent.jsonXContent.createParser(
-            NamedXContentRegistry.EMPTY,
-            LoggingDeprecationHandler.INSTANCE,
-            response.getEntity().getContent()
-        ).map();
-        List<Map<String, Object>> hits = (List<Map<String, Object>>) ((Map<String, Object>) responseJson.get("hits")).get("hits");
-        return hits.size() == 1;
     }
 }