feat(scheduler): Add multithreaded scheduler#73
Open
rawo wants to merge 17 commits into
Open
Conversation
…d optional virtual threads
Contributor
There was a problem hiding this comment.
Pull request overview
Adds configurable parallelism to the outbox scheduler so a single tick can execute multiple independent processNext() workers concurrently (disjoint row claims via FOR UPDATE SKIP LOCKED), and wires the knob through Spring Boot with tests/benchmarks/documentation.
Changes:
- Introduces
OutboxSchedulerConfig.concurrency(1–256) andworkerExecutorFactorywith platform/virtual thread pool helpers. - Updates
OutboxSchedulerto fan out work per tick and wait for all workers before the next interval. - Adds JMH benchmark + results writeup, updates docs, and extends integration/unit tests for concurrency behavior.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Documents scheduler concurrency benchmark results and tuning guidance. |
| okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt | Adds Spring property binding coverage for okapi.processor.concurrency and invalid-value startup failure. |
| okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt | Adds concurrency property with validation. |
| okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt | Wires Spring properties into OutboxSchedulerConfig(concurrency=...). |
| okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt | Adds integration test running real OutboxScheduler(concurrency=4) ensuring no amplification and disjoint claims. |
| okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt | Adds unit tests for fan-out, zero-overhead concurrency=1, and worker exception isolation. |
| okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt | Adds validation tests and executor factory behavior tests. |
| okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt | Adds concurrency + executor factory to scheduler config and companion factories. |
| okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt | Implements per-tick worker fan-out and worker pool shutdown. |
| okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt | Adds JMH benchmark for scheduler fan-out using platform vs virtual executors. |
| benchmarks/scheduler-concurrency.json | Stores raw JMH results for the new benchmark. |
| benchmarks/results-postopt-KOJAK-77.md | Adds benchmark analysis/writeup for KOJAK-77. |
| benchmarks/README.md | Documents the new benchmark and interpretation caveats. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…d optional virtual threads
…duler' into feat/kojak-77-multithreaded-scheduler # Conflicts: # okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt # okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt # okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt
Signed-off-by: Rafał Wokacz <rafal.wokacz@gmail.com>
…urency benchmarks tests
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt:41
- defaultPlatformPool(n) is a public factory, but it doesn't validate n. Passing 0 or negative will fail inside Executors with a less clear error; consider validating upfront for a more predictable API contract.
fun defaultPlatformPool(n: Int): ExecutorService {
require(n > 0) { "n must be positive, got: $n" }
val counter = AtomicInteger(0)
return Executors.newFixedThreadPool(n) { r ->
Thread(r, "outbox-worker-${counter.incrementAndGet()}").apply { isDaemon = true }
…s at OutboxSchedulerTest.kt
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (1)
okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt:150
- The WARN message in the shutdown interrupt path says "will retry at next scheduled interval", but this branch only runs when
runningis already false (i.e., duringstop()), so there is no next scheduled interval. This makes shutdown logs misleading for operators.
logger.warn("Interrupted while awaiting outbox worker; will retry at next scheduled interval", e)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Rafał Wokacz <rafal.wokacz@gmail.com>
Comment on lines
+146
to
+163
| private fun awaitWorker(future: Future<*>) { | ||
| while (true) { | ||
| try { | ||
| future.get() | ||
| return | ||
| } catch (e: InterruptedException) { | ||
| if (!running.get()) { | ||
| Thread.currentThread().interrupt() | ||
| logger.warn("Interrupted while awaiting outbox worker during shutdown", e) | ||
| return | ||
| } | ||
| // Not shutting down: keep waiting instead of letting the next tick overlap this worker. | ||
| } catch (e: ExecutionException) { | ||
| logger.error("Outbox worker failed unexpectedly, will retry at next scheduled interval", e.cause ?: e) | ||
| return | ||
| } | ||
| } | ||
| } |
Comment on lines
+85
to
+95
| private fun shutdownAndAwait(executor: ExecutorService) { | ||
| executor.shutdown() | ||
| try { | ||
| if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { | ||
| executor.shutdownNow() | ||
| } | ||
| } catch (_: InterruptedException) { | ||
| Thread.currentThread().interrupt() | ||
| executor.shutdownNow() | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements KOJAK-77: adds a
concurrencyknob toOutboxSchedulerConfig, letting a single scheduler tick fan out to N parallel workers instead of processing one batch at a time. Each worker claims its own disjoint batch via the existingFOR UPDATE SKIP LOCKED, so no app-level coordination is needed — cross-worker and cross-instance safety is free.OutboxSchedulerConfig: newconcurrency: Int(1-256, validated) andworkerExecutorFactory: (Int) -> ExecutorService, withdefaultPlatformPool(namedoutbox-worker-Ndaemon threads) andvirtualThreadPoolcompanion factory shortcuts.OutboxScheduler: fans out each tick toconcurrencyworkers via the configured executor, blocking until all complete before the next interval (so ticks never overlap).concurrency=1(default) skips the executor entirely — zero overhead, byte-for-byte the original single-worker behavior.okapi-spring-boot:okapi.processor.concurrencyproperty wired through with the same validation/startup-failure behavior as the existing processor properties.Benchmark & key finding
New
OutboxSchedulerConcurrencyBenchmark(JMH) measuresexecutorType ∈ {platform, virtual}×concurrency ∈ {1, 4, 16, 64}against Kafka'sdeliverBatchtransport (HTTPdeliverBatchisn't implemented yet — KOJAK-74 — so this ticket, formally blocked on it, benchmarks against Kafka instead).The ticket's virtual-thread hypothesis did not hold, and is reported as such rather than adjusted to fit: virtual threads showed no advantage over platform threads at any tested concurrency, even on JDK 25 (JEP 491, so the usual "pre-JDK-24 pinning" explanation doesn't apply) — platform was tied-or-ahead throughout. Full writeup, including why (workload never oversubscribes the platform pool at these sizes), the round-count artifact at concurrency=64, and the tuning recommendation:
benchmarks/results-postopt-KOJAK-77.md.Test plan
OutboxSchedulerConfigTest(concurrency validation 1-256,defaultPlatformPool/virtualThreadPoolfactory behavior),OutboxSchedulerTest(fan-out dispatches exactlyconcurrencyworkers per tick,concurrency=1never touchesworkerExecutorFactory, one worker's exception doesn't block the others)ConcurrentClaimTestsextended with a realOutboxScheduler(concurrency=4)against both Postgres and MySQL — disjoint claims across workers and ticks, zero delivery amplificationokapi-spring-boot: property binding + invalid-value startup-failure tests./gradlew ktlintCheckclean, full./gradlew buildgreenNot done / deliberately out of scope
processNext()'s single transaction to shorten lock hold time during delivery I/O — tracked separately as KOJAK-79 (decision-pending; needs actual lock-contention metrics from a high-concurrency run before deciding implement-vs-reject, which this PR's benchmark didn't specifically capture)Refs: KOJAK-77 (blocked-by KOJAK-73 ✅ done, KOJAK-74 ⏳ To Do; blocks KOJAK-79)