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

[mkuchenbecker]

Optimizer Stack

PR	Content
#527	Data Model
#530	Database Repos
#531	REST service
#533	Analyzer app
#534 (this)	Scheduler app
#tbd	Spark BatchedOFD app
#tbd	Infra, docker-compose, smoke test

Summary

PR 4 of N in the optimizer stack.

Introduces apps/optimizer-scheduler, a Spring Boot CommandLineRunner that claims PENDING operations and submits batched Spark jobs via the Jobs Service.

State machine:
Analyzer creates all Operations as PENDING

1. Scheduler marks PENDING as SCHEDULING to reserve the operation for a Bin ahead of scheduling to reduce duplicate job submission.
2. After claiming, the bin is submitted as a single job. If the batch is successfully submitted, the existing operations are updated to a status of scheduled with the associated jobID persisted.
3. If the submission fails, the job will be reset to pending so another scheduler run can pick it up.
4. Duplicate pending jobs are cancelled.
5. SCHEDULING is not expected to be in that state for long, minutes at most, and any operations that are stuck in the SCHEDULING status for >T period of time should move to a state of CANCELLED so next analyzer iteration will create a new PENDING op. This might happen if the scheduler was to suddenly crash or pod rotation. It may or may not happened after the job is scheduled. Therefore its safest to transition to CANCELLED in case the job was successful but not transitioned to scheduled. This is as-compared to failing to submit a job, in which case we know the job will never complete.

Changes

• Client-facing API Changes
• Internal API Changes
• Bug Fixes
• New Features
• Performance Improvements
• Code Style
• Refactoring
• Documentation
• Tests

Scheduler runner: Loads PENDING ops, bin-packs by file count, claims via two-step CAS (PENDING → SCHEDULING → SCHEDULED), submits one Spark job per bin.

Bin packer: Greedy first-fit descending algorithm. Oversized tables get their own bin (never dropped). Tables with no stats default to cost 0.

Jobs client: WebClient-based REST client submitting POST /jobs to the Jobs Service with table names, operation IDs, and results endpoint.

Repository additions: Three @Modifying CAS methods on TableOperationsRepository — cancelDuplicatePending, markScheduling, markScheduled — required for safe concurrent scheduling.

Testing Done

• Manually Tested on local docker setup. Please include commands ran, and their output.
• Added new tests for the changes made.
• Updated existing tests to reflect the changes made.
• No tests added or updated. Please explain why. If unsure, please feel free to ask for help.
• Some other form of testing like staging or soak time in production. Please explain.

13 unit tests:

• BinPackerTest (7 tests) — empty input, single table, under/over limit, oversized table, no stats, descending sort
• SchedulerRunnerTest (6 tests) — no pending ops, two-step claim + schedule, launch failure, already-claimed skip, duplicate cancellation, multi-row bin claim

./gradlew :apps:optimizer-scheduler:test
# BUILD SUCCESSFUL — 13 tests pass

Additional Information

• Breaking Changes
• Deprecations
• Large PR broken into smaller PRs, and PR plan linked in the description.

[mkuchenbecker]

I thought though this deeply:

Scheduler operation is as follows:
Scheduler s = new Scheduler().regiserOperation(Orphaned_file_deletion, ...).registerOperation(SE, ...)
s.schedule(Orphaned_File_deletion)

We register a FirstFitBinPacker bin packer for OFD or any operation that scales based on files size. There might be a singleton bin packer for operations we can't safely pack. Another could be based on table size, or take database into account.

The Bin Packer interface just cares that you can convert a TableStat to BinItem. By making a FirstFitBinPacker we are delegating the conversion to the generic type. The pattern is basically "A bin item is valid if if can be constructed from stats" and we can evolve that definition over time.

cc @abhisheknath2011

[abhisheknath2011]

Thanks @mkuchenbecker for addressing the comments!