Add indexing supervisor

[kfaraz]

Description

• Add a BatchIndexingSupervisor as an analog to the SeekableStreamSupervisor
• CompactionSupervisor implements BatchIndexingSupervisor
• ScheduledBatchSupervisor will also be updated to extend this supervisor in follow up PRs
• A BatchIndexingSupervisor uses BatchIndexingJobTemplate to create batch jobs (Task or MSQ SQL)

Changes

• Add BatchIndexingJob
• Add BatchIndexingSupervisor

Pending items

• fix code coverage
• fix up compaction stats collection
• store compaction state in MSQ jobs
• fill timeline gaps, support realiging intervals
• cancel mismatching task
• pass in the engine to the template
• Basic MSQ template
• invoke onTimelineUpdated - timeline will now get updated very frequently,
• we don't want to recompact intervals, try to find the right thing to do.
• we might have to do it via the policy
• maybe use searchInterval instead of skipIntervals
• how does this whole thing affect queuedIntervals
• for duty, it doesn't matter
• for supervisors, intervals will always be mutually exclusive

To fix the problem, remove the unused parameter P jobParams from the shouldCreateJobs method in the BatchIndexingSupervisor interface. This involves editing the method signature on line 38 to remove the parameter, so it becomes boolean shouldCreateJobs();. No other changes are needed in this file. However, all implementations of this interface and any code that calls this method will also need to be updated to remove the parameter, but since we are only allowed to edit the code shown, we will only update the interface here.

---

To fix the problem, we should replace the usage of the deprecated getMaxRowsPerSegment() method with its recommended alternative. Typically, when a method is deprecated, the Javadoc will suggest a replacement. In the context of Druid's compaction configuration, the replacement is likely getMaxRowsPerSegmentPerPartition() or a similar method, which provides more granular control. The fix should be applied only to the invocation at line 363 in server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java. No changes to imports are needed, as the replacement method should be available on the same class. The rest of the arguments to ClientCompactionTaskQueryTuningConfig.from() should remain unchanged to preserve existing functionality.

To fix the problem, we should add a case for the missing PENDING enum value in the switch statement on compactionStatus.getState() within the startJobIfPendingAndReady method. The best way to do this is to handle PENDING in a manner consistent with the method's logic. Since the method is about starting jobs that are pending and ready, and there is already logic for adding jobs to pending if not enough slots are available, it makes sense to treat PENDING similarly to SKIPPED (i.e., add to pending and return false), or to explicitly document why it should be handled differently. If the correct action is unclear, the safest approach is to add a case for PENDING that either handles it appropriately or throws an exception to make the code's intent clear.

The change should be made in the file indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java, specifically in the switch statement starting at line 213 in the startJobIfPendingAndReady method. No new imports or method definitions are required.