Scheduled batch supervisor

[abhishekrb19]

This change introduces a scheduled batch supervisor in Druid. The supervisor periodically wakes up to submit an MSQ ingest query, allowing users to automate batch ingestion directly within Druid. Think of it as simple batch task workflows natively integrated into Druid, though it doesn't replace more sophisticated workflow management systems like Apache Airflow. This is an experimental feature.

Summary of changes:

The scheduled_batch supervisor can be configured as follows:

{
    "type": "scheduled_batch",
    "schedulerConfig": {
        "type": "unix",
        "schedule": "*/5 * * * *"
    },
    "spec": {
        "query": "REPLACE INTO foo OVERWRITE ALL SELECT * FROM bar PARTITIONED BY DAY"
    },
    "suspended": false
}

The supervisor will submit the REPLACE sql query repeatedly every 5 minutes. The supervisor supports two types of cron scheduler configurations:

1. Unix cron syntax:
   • Type must be set to unix.
   • Follows the standard Unix cron format, e.g., */5 * * * * to schedule the SQL task every 5 minutes.
   • Supports macro expressions such as @daily, @hourly, @monthly, etc.
2. Quartz cron syntax:
   • Type must be set to quartz.
   • Offers more flexibility and control for scheduling tasks.
   • Example: 0 0 0 ? 3,6,9,12 MON-FRI to schedule tasks at midnight on weekdays during March, June, September, and December.

Key points:

• User can specify the query along with any context in the spec. This structure is identical to what the MSQ task engine accepts.
• Currently, the batch supervisor will repeatedly submit the DML SQL present in the spec as-is on its schedule.
• This can be useful for recurring batch ingestion or reindexing tasks.
• There is no parameterization or "change detection" support for input sources yet, but some simple mechanisms are planned for the future.
• Users can configure multiple scheduled batch supervisors for the same datasource.
• Only DML ingestion queries are currently supported by the scheduled batch supervisor.
• Users can suspend, resume and terminate the scheduled batch supervisor. Suspending the supervisor stops new tasks from being scheduled.

Some implementation details:

• The scheduled batch supervisor runs on the Overlord and communicates with the Broker to:
  a. Validate and parse the user-specified query.
  b. Submit MSQ queries to the /druid/v2/sql/task/ endpoint.
• A ScheduledBatchScheduler is injected in the Overlord, which is responsible for scheduling and unscheduling all scheduled batch instances.
• A BrokerClient implementation has been added, leveraging the ServiceClient functionality.
• The SqlTaskStatus and its unit test SqlTaskStatusTest have been moved from the msq module to the sql module so it can be reused by the BrokerClient implementation in the sql module.
• Added ExplainPlanInformation class, which is used to deserialize the explain plan response from the Broker.

The status API response for the supervisor contains the scheduler state along with active and completed tasks:

curl -X GET http://localhost:8888/druid/indexer/v1/supervisor/scheduled_batch__foo__3a916bcf-060d-4a42-aab6-904ff61d25cd/status

{
  "supervisorId": "scheduled_batch__foo__3764d545-62b8-4d52-a20c-568b009268e1",
  "state": "RUNNING",
  "lastTaskSubmittedTime": "2025-02-11T01:39:00.119Z",
  "nextTaskSubmissionTime": "2025-02-11T01:40:00.000Z",
  "taskReport": {
    "totalSubmittedTasks": 5,
    "totalSuccessfulTasks": 3,
    "totalFailedTasks": 2,
    "recentTasks": [
      {
        "taskStatus": {
          "id": "query-bb0fc147-2cde-4935-95ed-bf40a26568de",
          "status": "SUCCESS",
          "duration": 26628,
          "errorMsg": null,
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:37:26.780Z"
      },
      {
        "taskStatus": {
          "id": "query-1d51d1fc-ece5-4376-b64d-716ccecbd708",
          "status": "SUCCESS",
          "duration": 26367,
          "errorMsg": null,
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:38:26.493Z"
      },
      {
        "taskStatus": {
          "id": "query-baa8d243-ff57-4c18-b513-865687dd3e93",
          "status": "FAILED",
          "duration": 26197,
          "errorMsg": "Task execution process exited unsuccessfully with code[2]. See middleManager logs for more details.",
          "location": {
            "host": null,
            "port": -1,
            "tlsPort": -1
          }
        },
        "updatedTime": "2025-02-11T01:39:26.355Z"
      }
    ]
  }
}

Future Improvements:

• Scheduler State Persistence: This feature doesn't yet persist the scheduler state. Adding persistence will improve robustness (e.g., preventing missed jobs during Overlord restarts).
• Task Limits: There is currently no limit on the maximum number of tasks allowed by the scheduled batch supervisor. We may want to consider this in the future.
• The Supervisors page in the Druid web-console should change a bit to accommodate this change (more broadly non-streaming supervisors).

Release Note

This change introduces an experimental feature called scheduled batch supervisor in Druid. The supervisor periodically wakes up to submit an MSQ ingest query, allowing users to automate batch ingestion directly within Druid. Think of it as simple batch task workflows natively integrated into Druid, though it doesn't replace more sophisticated workflow management systems like Apache Airflow. Currently, the supervisor will repeatedly submit the same MSQ query as-is. The payload for the supervisor is below, please check the PR summary for additional details:

{
    "type": "scheduled_batch",
    "schedulerConfig": {
        "type": "unix",
        "schedule": "*/5 * * * *"
    },
    "spec": {
        "query": "REPLACE INTO foo OVERWRITE ALL SELECT * FROM bar PARTITIONED BY DAY"
    },
    "suspended": false
}

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[kfaraz]

Sorry for the delay on this, @abhishekrb19 !

I have tried to do a complete review. Once we have addressed these comments, we should be good to merge this PR.

Thanks a lot for your patience!! 🙂

[abhishekrb19]

@kfaraz, thank you for the reviews! I believe I have addressed and/or responded to all of your comments.

[kfaraz]

Minor non-blocking comments. 🚀

[kfaraz]

Nit: Since this constructor is not used for deserializing right now, we don't need the @JsonProperty annotation.

[kfaraz]

Nit: I am not sure if markdown is honored in javadocs. Use <i>, <b> or caps instead for emphasis.

[kfaraz]

@abhishekrb19 , the changes look good to me. You may proceed with the merge.

When you get the time though, please take another look at this comment
#17353 (comment)
If you feel that it's valid, it can be addressed in a follow up PR.

[abhishekrb19]

Sounds good, just responded in the comment inline