Enhance index monitor to terminate streaming job on consecutive errors

[dai-chen]

Description

This PR introduces enhancements to the FlintSparkIndexMonitor to improve the robustness and observability of index monitoring tasks. By tracking consecutive errors, this update helps in better managing the streaming tasks and resource utilization.

1. Add new Spark conf for monitor initial delay, interval and max error count
2. Add FlintSparkIndexMonitorTask with counter tracking number of consecutive errors
3. Add streaming job and monitor task terminating logic once max error count reached

Testing with EMR and AOS:

scala> sc.setLogLevel("INFO")

# Start index refresh streaming job
sql("CREATE SKIPPING INDEX ON myglue.ds_tables.http_logs_perf_tiny (status VALUE_SET) WITH (auto_refresh = true)")

24/05/17 21:28:13 INFO FlintSparkIndexMonitor: Starting index monitor for flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index with configuration:
 - Initial delay: 15 seconds
 - Interval: 60 seconds
 - Max error count: 5

# Set read only on AOS
PUT _cluster/settings
{
   "persistent":{
      "cluster.blocks.read_only": true
   }
}

# Monitor start failing
24/05/17 21:28:24 ERROR FlintSparkIndexMonitor: Failed to update index log entry, consecutive errors: 1

24/05/17 21:29:24 ERROR FlintSparkIndexMonitor: Failed to update index log entry, consecutive errors: 2
...

# Remove read only on AOS
PUT _cluster/settings
{
   "persistent":{
      "cluster.blocks.read_only": false
   }
}

# Monitor can update successfully
24/05/17 21:30:24 INFO FlintSparkIndexMonitor: Scheduler trigger index monitor task for flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index
24/05/17 21:30:24 INFO FlintSparkIndexMonitor: Streaming job is still active

# Set read only on AOS again
PUT _cluster/settings
{
   "persistent":{
      "cluster.blocks.read_only": true
   }
}

# Counter is reset and restarts from 1
24/05/17 21:31:25 ERROR FlintSparkIndexMonitor: Failed to update index log entry, consecutive errors: 1

24/05/17 21:32:25 ERROR FlintSparkIndexMonitor: Failed to update index log entry, consecutive errors: 2

...

# Monitor stops streaming job and itself once max error count reached
24/05/17 21:35:26 ERROR FlintSparkIndexMonitor: Failed to update index log entry, consecutive errors: 5
java.lang.IllegalStateException: Failed to commit transaction operation
	at org.opensearch.flint.core.metadata.log.DefaultOptimisticTransaction.commit(DefaultOptimisticTransaction.java:125) ~[flint-spark-integration-assembly-0.5.0-SNAPSHOT.jar:0.5.0-SNAPSHOT]
	at org.opensearch.flint.spark.FlintSparkIndexMonitor$FlintSparkIndexMonitorTask.run(FlintSparkIndexMonitor.scala:106) ~[flint-spark-integration-assembly-0.5.0-SNAPSHOT.jar:0.5.0-SNAPSHOT]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[?:?]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]
24/05/17 21:35:26 INFO FlintSparkIndexMonitor: Terminating streaming job and index monitor for flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index
24/05/17 21:35:26 INFO DAGScheduler: Asked to cancel job group 02c0af66-b3a9-4611-8fc2-9259206f83a4
24/05/17 21:35:26 ERROR MicroBatchExecution: Query flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index [id = ba336769-f33d-4da8-8ba2-46630cbdcd92, runId = 02c0af66-b3a9-4611-8fc2-9259206f83a4] terminated with error ...
24/05/17 21:35:26 INFO DAGScheduler: Asked to cancel job group 02c0af66-b3a9-4611-8fc2-9259206f83a4
24/05/17 21:35:26 INFO MicroBatchExecution: Query flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index [id = ba336769-f33d-4da8-8ba2-46630cbdcd92, runId = 02c0af66-b3a9-4611-8fc2-9259206f83a4] was stopped
24/05/17 21:35:26 INFO FlintSparkIndexMonitor: Cancelling scheduled task for index flint_myglue_ds_tables_http_logs_perf_tiny_skipping_index
24/05/17 21:35:26 INFO FlintSparkIndexMonitor: Streaming job and index monitor terminated

Issues Resolved

#344

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.