Scheduled batch supervisor

indexing-service/pom.xml

@@ -187,6 +187,11 @@
             <artifactId>commons-collections4</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.cronutils</groupId>
+            <artifactId>cron-utils</artifactId>
+            <version>${cronutils.version}</version>
+        </dependency>
         <!-- Tests -->
         <dependency>
             <groupId>junit</groupId>

indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java

@@ -27,6 +27,7 @@
 import com.google.inject.Binder;
 import org.apache.druid.indexing.compact.CompactionSupervisorSpec;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.scheduledbatch.ScheduledBatchSupervisorSpec;
 import org.apache.druid.initialization.DruidModule;
 import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
 
@@ -47,7 +48,8 @@ public List<? extends Module> getJacksonModules()
     return ImmutableList.of(
         new SimpleModule(getClass().getSimpleName())
             .registerSubtypes(
-                new NamedType(CompactionSupervisorSpec.class, CompactionSupervisorSpec.TYPE)
+                new NamedType(CompactionSupervisorSpec.class, CompactionSupervisorSpec.TYPE),
+                new NamedType(ScheduledBatchSupervisorSpec.class, ScheduledBatchSupervisorSpec.TYPE)
             )
     );
   }

indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java

@@ -35,6 +35,7 @@
 import org.apache.druid.indexing.overlord.config.TaskQueueConfig;
 import org.apache.druid.indexing.overlord.duty.OverlordDutyExecutor;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import org.apache.druid.indexing.scheduledbatch.ScheduledBatchTaskManager;
 import org.apache.druid.java.util.common.lifecycle.Lifecycle;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
@@ -89,6 +90,7 @@ public DruidOverlord(
       @IndexingService final DruidLeaderSelector overlordLeaderSelector,
       final SegmentAllocationQueue segmentAllocationQueue,
       final CompactionScheduler compactionScheduler,
+      final ScheduledBatchTaskManager scheduledBatchTaskManager,
       final ObjectMapper mapper,
       final TaskContextEnricher taskContextEnricher
   )
@@ -161,6 +163,7 @@ public void start()
                 {
                   taskMaster.becomeFullLeader();
                   compactionScheduler.start();
+                  scheduledBatchTaskManager.start();
 
                   // Announce the node only after all the services have been initialized
                   initialized = true;
@@ -171,6 +174,7 @@ public void start()
                 public void stop()
                 {
                   serviceAnnouncer.unannounce(node);
+                  scheduledBatchTaskManager.stop();
                   compactionScheduler.stop();
                   taskMaster.downgradeToHalfLeader();
                 }

indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/BatchSupervisorTaskReport.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.scheduledbatch;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+
+/**
+ * Represents a summary report of task execution for a batch supervisor, including total counts
+ * of submitted, successful, and failed tasks, along with a recent set of tasks.
+ */
+public class BatchSupervisorTaskReport
+{
+  private final int totalSubmittedTasks;
+  private final int totalSuccessfulTasks;
+  private final int totalFailedTasks;
+  private final List<BatchSupervisorTaskStatus> recentTasks;
+
+  public BatchSupervisorTaskReport(
+      @JsonProperty("totalSubmittedTasks") final int totalSubmittedTasks,
+      @JsonProperty("totalSuccessfulTasks") final int totalSuccessfulTasks,
+      @JsonProperty("totalFailedTasks") final int totalFailedTasks,
+      @JsonProperty("recentTasks") final List<BatchSupervisorTaskStatus> recentTasks
+  )
+  {
+    this.totalSubmittedTasks = totalSubmittedTasks;
+    this.totalSuccessfulTasks = totalSuccessfulTasks;
+    this.totalFailedTasks = totalFailedTasks;
+    this.recentTasks = recentTasks;
+  }
+
+  @JsonProperty
+  public int getTotalSubmittedTasks()
+  {
+    return totalSubmittedTasks;
+  }
+
+  @JsonProperty
+  public int getTotalSuccessfulTasks()
+  {
+    return totalSuccessfulTasks;
+  }
+
+  @JsonProperty
+  public int getTotalFailedTasks()
+  {
+    return totalFailedTasks;
+  }
+
+  @JsonProperty
+  public List<BatchSupervisorTaskStatus> getRecentTasks()
+  {
+    return recentTasks;
+  }
+}

indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/BatchSupervisorTaskStatus.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.scheduledbatch;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.indexer.TaskStatus;
+import org.joda.time.DateTime;
+
+/**
+ * Represents the status of a scheduled batch supervisor task and the timestamp of its last update.
+ */
+public class BatchSupervisorTaskStatus
+{
+  private final String supervisorId;
+  private final TaskStatus taskStatus;
+  private final DateTime updatedTime;
+
+  public BatchSupervisorTaskStatus(
+      String supervisorId, // This field is used only for internal tracking, so not Jackson serializable
+      @JsonProperty("taskStatus") TaskStatus taskStatus,
+      @JsonProperty("updatedTime") DateTime updatedTime
+  )
+  {
+    this.supervisorId = supervisorId;
+    this.taskStatus = taskStatus;
+    this.updatedTime = updatedTime;
+  }
+
+  /**
+   * Used for internal tracking. So this field is *not* Jackson serialized to avoid
+   * redundant information in the user-facing objects.
+   */
+  @JsonIgnore
+  public String getSupervisorId()
+  {
+    return supervisorId;
+  }
+
+  @JsonProperty
+  public TaskStatus getTaskStatus()
+  {
+    return taskStatus;
+  }
+
+  @JsonProperty
+  public DateTime getUpdatedTime()
+  {
+    return updatedTime;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "BatchSupervisorTaskStatus{" +
+           "supervisorId='" + supervisorId + '\'' +
+           ", taskStatus=" + taskStatus +
+           ", updatedTime=" + updatedTime +
+           '}';
+  }
+}

indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/CronSchedulerConfig.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.scheduledbatch;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+
+import javax.annotation.Nullable;
+
+/**
+ * Interface representing a configuration for scheduling tasks based on cron expressions.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = UnixCronSchedulerConfig.TYPE, value = UnixCronSchedulerConfig.class),
+    @JsonSubTypes.Type(name = QuartzCronSchedulerConfig.TYPE, value = QuartzCronSchedulerConfig.class),
+})
+public interface CronSchedulerConfig
+{
+  /**
+   * Gets the next task submission time after the specified {@code referenceTime}, or {@code null} if no future
+   * execution time exists based on the configured schedule (e.g., if the cron expression only specifies dates in the past).
+   */
+  @Nullable
+  DateTime getNextTaskStartTimeAfter(DateTime referenceTime);
+
+  /**
+   * Gets the duration until the next task submission after the specified {@code referenceTime}, if no future
+   * execution time exists based on the configured schedule (e.g., if the cron expression only specifies dates in the past).
+   */
+  @Nullable
+  Duration getDurationUntilNextTaskStartTimeAfter(DateTime referenceTime);
+}

indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/CronSchedulerUtils.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.scheduledbatch;
+
+import com.cronutils.model.time.ExecutionTime;
+import org.apache.druid.java.util.common.DateTimes;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+
+import javax.annotation.Nullable;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.Optional;
+
+public class CronSchedulerUtils
+{
+  /**
+   * Computes the next task submission time after the specified {@code referenceTime}.
+   * <p>
+   * Returns {@code null} if no future execution time exists based on the given {@code executionTime},
+   * which can occur if the schedule does not define any future occurrences.
+   * </p>
+   */
+  @Nullable
+  static DateTime getNextTaskStartTimeAfter(final ExecutionTime executionTime, final DateTime referenceTime)
+  {
+    final Optional<ZonedDateTime> zonedDateTime = executionTime.nextExecution(convertToZonedDateTime(referenceTime));
+    if (zonedDateTime.isPresent()) {
+      final ZonedDateTime zdt = zonedDateTime.get();
+      final Instant instant = zdt.toInstant();
+      return new DateTime(instant.toEpochMilli(), DateTimes.inferTzFromString(zdt.getZone().getId()));
+    } else {
+      return null;
+    }
+  }
+
+  /**
+   * Computes the duration until the next task submission time after the specified {@code referenceTime}.
+   * <p>
+   * Returns {@code null} if no future execution time exists based on the given {@code executionTime},
+   * which can occur if the schedule does not define any future occurrences.
+   * </p>
+   */
+  @Nullable
+  static Duration getDurationUntilNextTaskStartTimeAfter(final ExecutionTime executionTime, final DateTime referenceTime)
+  {
+    final Optional<java.time.Duration> duration = executionTime.timeToNextExecution(convertToZonedDateTime(referenceTime));
+    return duration.map(value -> Duration.millis(value.toMillis())).orElse(null);
+  }
+
+  private static ZonedDateTime convertToZonedDateTime(final DateTime jodaDateTime)
+  {
+    return ZonedDateTime.ofInstant(
+        Instant.ofEpochMilli(jodaDateTime.getMillis()),
+        ZoneId.of(jodaDateTime.getZone().getID())
+    );
+  }
+}