feat(optimizer): [4/N] Scheduler app

apps/optimizer/schedulerapp/build.gradle

@@ -0,0 +1,15 @@
+plugins {
+  id 'openhouse.springboot-ext-conventions'
+  id 'openhouse.maven-publish'
+  id 'org.springframework.boot' version '2.7.8'
+}
+
+// Deployable Spring Boot wrapper around the scheduler library. Holds SchedulerApplication (the
+// @SpringBootApplication entry point) and application.properties; the scheduling logic lives in
+// :services:optimizer:scheduler.
+dependencies {
+  implementation project(':services:optimizer:scheduler')
+  implementation 'org.springframework.boot:spring-boot-starter:2.7.8'
+  implementation 'org.springframework.boot:spring-boot-starter-data-jpa:2.7.8'
+  runtimeOnly 'mysql:mysql-connector-java:8.0.33'
+}

apps/optimizer/schedulerapp/src/main/java/com/linkedin/openhouse/optimizer/scheduler/SchedulerApplication.java

@@ -0,0 +1,59 @@
+package com.linkedin.openhouse.optimizer.scheduler;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.ExitCodeGenerator;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+
+/**
+ * Entry point for the Optimizer Scheduler application.
+ *
+ * <p>Spring Batch–style: implements {@link CommandLineRunner} so the work runs after context
+ * startup, and {@link ExitCodeGenerator} so the JVM exit code reflects batch outcome. {@code
+ * SpringApplication.exit(...)} closes the context (triggers {@code @PreDestroy} hooks, drains the
+ * JPA pool, etc.) so the k8s CronJob pod terminates cleanly with a status reflecting reality.
+ */
+@Slf4j
+@SpringBootApplication
+@EntityScan(basePackages = "com.linkedin.openhouse.optimizer.db")
+@EnableJpaRepositories(basePackages = "com.linkedin.openhouse.optimizer.repository")
+public class SchedulerApplication implements CommandLineRunner, ExitCodeGenerator {
+
+  private final SchedulerRunner runner;
+  private int exitCode = 0;
+
+  @Autowired
+  public SchedulerApplication(SchedulerRunner runner) {
+    this.runner = runner;
+  }
+
+  public static void main(String[] args) {
+    System.exit(SpringApplication.exit(SpringApplication.run(SchedulerApplication.class, args)));
+  }
+
+  /**
+   * Runs the scheduler once per registered operation type per process invocation. Any thrown
+   * exception is logged and surfaces as a non-zero exit code via {@link #getExitCode()} after the
+   * context is shut down cleanly.
+   */
+  @Override
+  public void run(String... args) {
+    try {
+      log.info("Scheduler starting; operation types: {}", runner.getRegisteredOperationTypes());
+      runner.getRegisteredOperationTypes().forEach(runner::schedule);
+      log.info("Scheduler completed successfully");
+    } catch (Exception e) {
+      log.error("Scheduler failed", e);
+      exitCode = 1;
+    }
+  }
+
+  @Override
+  public int getExitCode() {
+    return exitCode;
+  }
+}

apps/optimizer/schedulerapp/src/main/resources/application.properties

@@ -0,0 +1,14 @@
+spring.application.name=openhouse-optimizer-scheduler
+spring.main.web-application-type=none
+spring.main.banner-mode=off
+spring.datasource.url=${OPTIMIZER_DB_URL:jdbc:h2:mem:schedulerdb;DB_CLOSE_DELAY=-1;MODE=MySQL}
+spring.datasource.username=${OPTIMIZER_DB_USER:sa}
+spring.datasource.password=${OPTIMIZER_DB_PASSWORD:}
+spring.jpa.hibernate.ddl-auto=none
+optimizer.scheduler.jobs.base-uri=${JOBS_BASE_URI:http://localhost:8002}
+# Per-bin caps for ORPHAN_FILES_DELETION; 0 disables a dimension. File count is the OFD cost
+# driver — per-file list, manifest joins, and delete calls dominate, independent of file size.
+optimizer.scheduler.ofd.max-files-per-bin=${SCHEDULER_OFD_MAX_FILES_PER_BIN:1000000}
+optimizer.scheduler.ofd.max-tables-per-bin=${SCHEDULER_OFD_MAX_TABLES_PER_BIN:50}
+optimizer.scheduler.results-endpoint=${SCHEDULER_RESULTS_ENDPOINT:http://openhouse-optimizer:8080/v1/optimizer/operations}
+optimizer.scheduler.cluster-id=${SCHEDULER_CLUSTER_ID:LocalHadoopCluster}

libs/optimizer/binpack/build.gradle

@@ -0,0 +1,15 @@
+plugins {
+  id 'openhouse.java-conventions'
+  id 'openhouse.maven-publish'
+}
+
+dependencies {
+  // api: bin-packing types reference TableOperationDto / TableStatsDto, which consumers need on
+  // their compile classpath.
+  api project(':services:optimizer')
+  api 'org.slf4j:slf4j-api:1.7.36'
+}
+
+test {
+  useJUnitPlatform()
+}

libs/optimizer/binpack/src/main/java/com/linkedin/openhouse/optimizer/binpack/Bin.java

@@ -0,0 +1,20 @@
+package com.linkedin.openhouse.optimizer.binpack;
+
+import com.linkedin.openhouse.optimizer.model.OperationTypeDto;
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ * One scheduling unit: the operation type the bin will run as, and the items the scheduler will
+ * claim, narrow to claimed, and launch a single Spark job for. Pure data — the scheduler reads from
+ * a bin to do the work; the bin does no IO itself.
+ */
+@AllArgsConstructor
+@Getter
+@ToString
+public class Bin {
+  private final OperationTypeDto operationType;
+  private final List<BinItem> items;
+}

libs/optimizer/binpack/src/main/java/com/linkedin/openhouse/optimizer/binpack/BinItem.java

@@ -0,0 +1,24 @@
+package com.linkedin.openhouse.optimizer.binpack;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsDto;
+
+/**
+ * One packable unit. Exposes the weight a packer keys on, plus the identity the scheduler reads
+ * when it launches a Spark job (fully-qualified table name, operation id).
+ *
+ * <p>Implementations have a public no-arg constructor — instantiated transiently inside {@link
+ * FirstFitBinPacker#pack} via a {@code Supplier<T extends BinItem>} (typically a {@code
+ * MyItem::new} method reference) — on which {@link #fromOpAndStats} is called to return the
+ * populated item. Getters on the empty instance are not meaningful; it exists for the lifetime of a
+ * single projection call.
+ */
+public interface BinItem {
+  long getWeight();
+
+  String getFullyQualifiedTableName();
+
+  String getOperationId();
+
+  BinItem fromOpAndStats(TableOperationDto op, TableStatsDto stats);
+}

libs/optimizer/binpack/src/main/java/com/linkedin/openhouse/optimizer/binpack/BinPacker.java

@@ -0,0 +1,18 @@
+package com.linkedin.openhouse.optimizer.binpack;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsDto;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Per-operation-type strategy. Given a flat list of operations and the corresponding stats, returns
+ * one grouping per batch the scheduler should submit. The scheduler wraps each grouping into a
+ * {@link Bin} with the registered operation type. Implementations do no IO and hold no mutable
+ * state; the projection from {@code (op, stats)} to {@link BinItem} and the bucketing strategy both
+ * live in the implementation.
+ */
+public interface BinPacker {
+  List<List<BinItem>> pack(
+      List<TableOperationDto> operations, Map<String, TableStatsDto> statsByTableUuid);
+}

libs/optimizer/binpack/src/main/java/com/linkedin/openhouse/optimizer/binpack/FirstFitBinPacker.java

@@ -0,0 +1,92 @@
+package com.linkedin.openhouse.optimizer.binpack;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsDto;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * First-fit-decreasing packing, generic over the concrete {@link BinItem} subtype {@code T}.
+ * Construction takes a {@code Supplier<T>} — typically a {@code MyItem::new} method reference —
+ * which the packer invokes per operation to get an empty instance, then calls {@link
+ * BinItem#fromOpAndStats} on it to project the (operation, stats) pair into a populated item.
+ *
+ * <p>Sorts items by weight descending, then places each into the first group whose totals stay at
+ * or below {@code maxWeightPerBin} and {@code maxItemsPerBin}. An item whose weight exceeds the cap
+ * on its own goes into a group by itself. Operations whose {@code tableUuid} has no entry in {@code
+ * statsByTableUuid} are dropped.
+ *
+ * <p>Stateless: the constructor takes only the BinItem supplier and the cap configuration; {@link
+ * #pack} is a pure function over its arguments. The packer is operation-agnostic — the scheduler
+ * wraps each grouping into a {@link Bin} with the registered operation type.
+ */
+@Slf4j
+public class FirstFitBinPacker<T extends BinItem> implements BinPacker {
+
+  private final Supplier<T> binItemSupplier;
+  private final long maxWeightPerBin;
+  private final int maxItemsPerBin;
+
+  public FirstFitBinPacker(Supplier<T> binItemSupplier, long maxWeightPerBin, int maxItemsPerBin) {
+    this.binItemSupplier = binItemSupplier;
+    this.maxWeightPerBin = maxWeightPerBin;
+    this.maxItemsPerBin = maxItemsPerBin;
+  }
+
+  @Override
+  public List<List<BinItem>> pack(
+      List<TableOperationDto> operations, Map<String, TableStatsDto> statsByTableUuid) {
+    List<BinItem> items =
+        operations.stream()
+            .filter(op -> statsByTableUuid.containsKey(op.getTableUuid()))
+            .map(
+                op ->
+                    binItemSupplier
+                        .get()
+                        .fromOpAndStats(op, statsByTableUuid.get(op.getTableUuid())))
+            .collect(Collectors.toList());
+    List<PackingBin> packingBins =
+        items.stream()
+            .sorted(Comparator.comparingLong(BinItem::getWeight).reversed())
+            .collect(ArrayList::new, this::placeItem, List::addAll);
+    log.info(
+        "Packed {} operations ({} items after projection) into {} groupings",
+        operations.size(),
+        items.size(),
+        packingBins.size());
+    return packingBins.stream().map(pb -> pb.items).collect(Collectors.toList());
+  }
+
+  private void placeItem(List<PackingBin> bins, BinItem item) {
+    bins.stream()
+        .filter(b -> b.fits(item, maxWeightPerBin, maxItemsPerBin))
+        .findFirst()
+        .ifPresentOrElse(
+            b -> b.add(item),
+            () -> {
+              PackingBin fresh = new PackingBin();
+              fresh.add(item);
+              bins.add(fresh);
+            });
+  }
+
+  /** Running-totals helper used during the fold. */
+  private static class PackingBin {
+    final List<BinItem> items = new ArrayList<>();
+    long totalWeight;
+
+    boolean fits(BinItem item, long maxWeight, int maxItems) {
+      return items.size() < maxItems && totalWeight + item.getWeight() <= maxWeight;
+    }
+
+    void add(BinItem item) {
+      items.add(item);
+      totalWeight += item.getWeight();
+    }
+  }
+}

libs/optimizer/binpack/src/main/java/com/linkedin/openhouse/optimizer/binpack/TotalFilesBinItem.java

@@ -0,0 +1,49 @@
+package com.linkedin.openhouse.optimizer.binpack;
+
+import com.linkedin.openhouse.optimizer.model.TableOperationDto;
+import com.linkedin.openhouse.optimizer.model.TableStatsDto;
+import java.util.Optional;
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ * {@link BinItem} that weights by the table's current file count. Suitable for any operation whose
+ * Spark cost scales with file count — orphan files deletion, stats collection, etc. The
+ * implementation knows nothing about which operation type it is wired up to.
+ *
+ * <p>Construction: callers pass {@code TotalFilesBinItem::new} as the {@code Supplier<T>} to {@link
+ * FirstFitBinPacker}; the packer calls the supplier per operation to get an empty instance, then
+ * {@link #fromOpAndStats} on it to get a populated copy.
+ */
+@Getter
+@ToString
+public class TotalFilesBinItem implements BinItem {
+
+  private final String fullyQualifiedTableName;
+  private final String operationId;
+  private final long weight;
+
+  /** Empty constructor: call {@link #fromOpAndStats} on the result to get a populated instance. */
+  public TotalFilesBinItem() {
+    this("", "", 0L);
+  }
+
+  private TotalFilesBinItem(String fullyQualifiedTableName, String operationId, long weight) {
+    this.fullyQualifiedTableName = fullyQualifiedTableName;
+    this.operationId = operationId;
+    this.weight = weight;
+  }
+
+  @Override
+  public BinItem fromOpAndStats(TableOperationDto op, TableStatsDto stats) {
+    return new TotalFilesBinItem(
+        op.getDatabaseName() + "." + op.getTableName(), op.getId(), currentFileCount(stats));
+  }
+
+  private static long currentFileCount(TableStatsDto stats) {
+    return Optional.ofNullable(stats)
+        .map(TableStatsDto::getSnapshot)
+        .map(TableStatsDto.SnapshotMetrics::getNumCurrentFiles)
+        .orElse(0L);
+  }
+}