Replace _id sort with _seq_no in JobSweeper to fix fielddata error

alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt

@@ -1559,4 +1559,31 @@ class MonitorRestApiIT : AlertingRestTestCase() {
             alertingStatsResponse[statsResponseOpenSearchSweeperEnabledField]
         )
     }
+
+    fun `test sweeper works with id field data disabled`() {
+        client().updateSettings(ScheduledJobSettings.SWEEPER_ENABLED.key, true)
+        val monitor = createRandomMonitor(refresh = true)
+        // Disable _id fielddata — this previously broke the sweeper
+        client().updateSettings("indices.id_field_data.enabled", false)
+        try {
+            val monitor2 = createRandomMonitor(refresh = true)
+            assertNotNull("Monitor was not created", monitor2.id)
+            val executeResponse = executeMonitor(monitor2.id)
+            assertEquals("Execute monitor failed", RestStatus.OK, executeResponse.restStatus())
+        } finally {
+            client().updateSettings("indices.id_field_data.enabled", true)
+        }
+    }
+
+    fun `test sweeper works after all monitors deleted`() {
+        client().updateSettings(ScheduledJobSettings.SWEEPER_ENABLED.key, true)
+        val monitor = createRandomMonitor(refresh = true)
+        client().makeRequest("DELETE", "$ALERTING_BASE_URI/${monitor.id}")
+        refreshIndex(ScheduledJob.SCHEDULED_JOBS_INDEX)
+        // Create a new monitor after index was emptied — sweeper should handle this
+        val monitor2 = createRandomMonitor(refresh = true)
+        assertNotNull("Monitor was not created", monitor2.id)
+        val executeResponse = executeMonitor(monitor2.id)
+        assertEquals("Execute monitor failed", RestStatus.OK, executeResponse.restStatus())
+    }
 }

core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt

@@ -295,7 +295,7 @@ class JobSweeper(
         lastFullSweepTimeNano = System.nanoTime()
     }
 
-    private fun sweepShard(shardId: ShardId, shardNodes: ShardNodes, startAfter: String = "") {
+    private fun sweepShard(shardId: ShardId, shardNodes: ShardNodes, startAfter: Long = -1L) {
         val logger = Loggers.getLogger(javaClass, shardId)
         logger.debug("Sweeping shard $shardId")
 
@@ -308,7 +308,7 @@ class JobSweeper(
 
         // sweep the shard for new and updated jobs. Uses a search after query to paginate, assuming that any concurrent
         // updates and deletes are handled by the index operation listener.
-        var searchAfter: String? = startAfter
+        var searchAfter: Long? = startAfter
         while (searchAfter != null) {
             val boolQueryBuilder = BoolQueryBuilder()
             sweepableJobTypes.forEach { boolQueryBuilder.should(QueryBuilders.existsQuery(it)) }
@@ -318,18 +318,23 @@ class JobSweeper(
                 .source(
                     SearchSourceBuilder.searchSource()
                         .version(true)
+                        .seqNoAndPrimaryTerm(true)
                         .sort(
-                            FieldSortBuilder("_id")
-                                .unmappedType("keyword")
-                                .missing("_last")
+                            FieldSortBuilder("_seq_no")
+                                .unmappedType("long")
                         )
                         .searchAfter(arrayOf(searchAfter))
                         .size(sweepPageSize)
                         .query(boolQueryBuilder)
                 )
 
-            val response = sweepSearchBackoff.retry {
-                client.search(jobSearchRequest).actionGet(requestTimeout)
+            val response = try {
+                sweepSearchBackoff.retry {
+                    client.search(jobSearchRequest).actionGet(requestTimeout)
+                }
+            } catch (e: Exception) {
+                logger.error("Aborting sweep of shard $shardId, will retry on next sweep cycle.", e)
+                return
             }
             if (response.status() != RestStatus.OK) {
                 logger.error("Error sweeping shard $shardId.", response.firstFailureOrNull())
@@ -344,7 +349,7 @@ class JobSweeper(
                     parseAndSweepJob(xcp, shardId, hit.id, hit.version, hit.sourceRef)
                 }
             }
-            searchAfter = response.hits.lastOrNull()?.id
+            searchAfter = response.hits.lastOrNull()?.seqNo
         }
     }