From 56b1e1f46c78feca7f5f2d81d2bfaff8197a78f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rafa=C5=82=20Wokacz?= Date: Fri, 17 Jul 2026 09:41:26 +0200 Subject: [PATCH] feat(scheduler): Add multithreaded scheduler with platform threads and optional virtual threads --- README.md | 26 +- benchmarks/README.md | 18 +- benchmarks/results-postopt-KOJAK-77.md | 104 ++++ benchmarks/scheduler-concurrency.json | 532 ++++++++++++++++++ .../OutboxSchedulerConcurrencyBenchmark.kt | 121 ++++ .../okapi/core/OutboxScheduler.kt | 38 +- .../okapi/core/OutboxSchedulerConfig.kt | 36 ++ .../okapi/core/OutboxSchedulerConfigTest.kt | 50 +- .../okapi/core/OutboxSchedulerTest.kt | 71 +++ .../test/concurrency/ConcurrentClaimTests.kt | 49 ++ .../springboot/OutboxAutoConfiguration.kt | 1 + .../springboot/OutboxProcessorProperties.kt | 6 + .../OutboxProcessorAutoConfigurationTest.kt | 11 + 13 files changed, 1053 insertions(+), 10 deletions(-) create mode 100644 benchmarks/results-postopt-KOJAK-77.md create mode 100644 benchmarks/scheduler-concurrency.json create mode 100644 okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt diff --git a/README.md b/README.md index 84bf051..281ac21 100644 --- a/README.md +++ b/README.md @@ -304,9 +304,29 @@ Throughput on a single instance (MacBook M3 Max, JDK 21 LTS, May 2026): | HTTP @ webhook latency 20 ms (sync sequential — parallel `sendAsync` planned) | ~38 msg/s | ~38 msg/s | | HTTP @ webhook latency 100 ms (sync sequential — parallel `sendAsync` planned) | ~9 msg/s | ~9 msg/s | -Kafka throughput jumped 16-45× over the original sync-sequential baseline thanks to the `deliverBatch` fire-flush-await pattern. HTTP parallel `sendAsync` is next; multi-threaded scheduler scaling is in the roadmap. - -Full methodology, raw JMH results, before/after per change: [`benchmarks/`](benchmarks/). +Kafka throughput jumped 16-45× over the original sync-sequential baseline thanks to the `deliverBatch` fire-flush-await pattern. HTTP parallel `sendAsync` is next. + +**Multi-threaded scheduler** (`OutboxSchedulerConfig.concurrency`, JDK 25, single Postgres+Kafka backend): + +| concurrency | speedup vs. concurrency=1 | +|---|---| +| 4 | **3.6×** | +| 16 | **6.2×** | +| 64 | **6.6×** (diminishing — see caveats below) | + +`concurrency=4` to `16` is the practical sweet spot — most of the available speedup is already +captured there, with marginal gains beyond it on a single-instance backend. Default to platform +threads (`workerExecutorFactory` default): in this benchmark's tested range (1-64 workers), +switching to `virtualThreadPool` showed **no measurable advantage** over platform threads, even +on a JEP 491 JDK (25) — contrary to the original hypothesis that virtual threads would win at +concurrency=16+. Virtual threads only pay off when worker count vastly exceeds the platform pool +size; at ≤64 workers there's no oversubscription for them to fix. Tuning rule of thumb: +`concurrency × instances ≤ max_connections / 2` (row locks make cross-instance coordination free +via `FOR UPDATE SKIP LOCKED`, but every worker holds a DB connection for its batch's duration). + +Full methodology, raw JMH results, before/after per change: [`benchmarks/`](benchmarks/), including +[`results-postopt-KOJAK-77.md`](benchmarks/results-postopt-KOJAK-77.md) for the full concurrency +breakdown and the reasoning behind the virtual-thread finding. ## Build diff --git a/benchmarks/README.md b/benchmarks/README.md index 1ddcadc..39b8ea8 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -67,6 +67,14 @@ Single-entry `deliver()` calls with mocked I/O: Measures pure code overhead (JSON deserialization, record/request construction, exception classification). Useful as "did optimization X regress the hot path?" baseline. +### Scheduler fan-out (`OutboxSchedulerConcurrencyBenchmark`) + +Drains a fixed backlog through `concurrency` parallel `processNext()` calls per round — one call +per worker, each on its own transaction, mirroring `OutboxScheduler`'s internal fan-out (the +scheduler's polling loop itself is still bypassed). `@Param executorType ∈ {platform, virtual}` × +`@Param concurrency ∈ {1, 4, 16, 64}`. Transport is Kafka `deliverBatch` — real Postgres + real +Kafka via Testcontainers. See [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md). + ## How to read results Throughput benchmarks report **ops/s = msg/s** thanks to `@OperationsPerInvocation`. @@ -96,10 +104,16 @@ investigate variability sources (background processes, thermal throttling, GC). is dominated by the target service's processing time + network — pick the value closest to your target. With `httpLatencyMs=100`, sequential delivery is bounded at `1000ms / 100ms = 10 msg/s/thread` regardless of library efficiency. -- **Single-threaded scheduler.** Current `OutboxSchedulerConfig` does not expose `concurrency`. - Once that lands (planned), the throughput matrix will expand to `batchSize × concurrency`. +- **Multi-threaded scheduler is single-instance, single-backend.** `OutboxSchedulerConcurrencyBenchmark` + fans out against one Postgres container and one Kafka broker; it measures the fan-out + mechanism's overhead, not unlimited horizontal DB/broker scaling. See + [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md) for the concurrency=64 round-count + artifact (a single round drains the whole fixed backlog, so per-round fixed costs aren't + amortized the way they are at lower concurrency). ## Historical baselines - [`results-baseline-2026-04.md`](results-baseline-2026-04.md) — pre-optimization baseline (sync sequential delivery, single-threaded scheduler) +- [`results-postopt-KOJAK-75.md`](results-postopt-KOJAK-75.md) — batch UPDATE via JDBC `executeBatch` +- [`results-postopt-KOJAK-77.md`](results-postopt-KOJAK-77.md) — multi-threaded scheduler concurrency knob diff --git a/benchmarks/results-postopt-KOJAK-77.md b/benchmarks/results-postopt-KOJAK-77.md new file mode 100644 index 0000000..2611edc --- /dev/null +++ b/benchmarks/results-postopt-KOJAK-77.md @@ -0,0 +1,104 @@ +# Multi-threaded scheduler with pluggable executor — Results (KOJAK-77) + +Measured on the same machine/config as [`results-kafka-deliverbatch.md`](results-kafka-deliverbatch.md): +Postgres 16 + Kafka 3.8.1 via Testcontainers, full JMH config `fork=2, warmup=3 × 10s, iter=5 × 30s` +(n=10 samples per benchmark). **JDK 25.0.2** this time (not the 21 LTS used for earlier docs in +this directory) — notable because it means [JEP 491](https://openjdk.org/jeps/491) (removes +`synchronized`-block carrier-thread pinning for virtual threads) is present; the "virtual threads +lose to pinning on older JDKs" excuse does not apply to this run. + +`OutboxSchedulerConcurrencyBenchmark` drains a fixed backlog of 6,400 pending entries through +`concurrency` parallel `OutboxProcessor.processNext(batchSize=100)` calls per round — one call per +worker, each on its own transaction/connection, exactly mirroring `OutboxScheduler`'s internal +fan-out (the scheduler's polling loop itself is bypassed, same rationale as the other throughput +benchmarks in this directory). Kafka `deliverBatch` is the transport, since HTTP `deliverBatch` +(KOJAK-74) isn't implemented yet and this benchmark is meant to isolate fan-out scaling, not +transport I/O. + +## Headline numbers + +| concurrency | platform (ms/op) | virtual (ms/op) | platform msg/s | virtual msg/s | virtual vs platform | +|---|---|---|---|---|---| +| 1 | 0.357 ± 0.097 | 0.347 ± 0.070 | ~2,801 | ~2,882 | +3% (noise — CIs overlap heavily) | +| 4 | 0.099 ± 0.003 | 0.104 ± 0.003 | ~10,101 | ~9,615 | **−5%** | +| 16 | 0.058 ± 0.005 | 0.058 ± 0.002 | ~17,241 | ~17,241 | 0% (tied) | +| 64 | 0.054 ± 0.002 | 0.058 ± 0.003 | ~18,519 | ~17,241 | **−7%** | + +Raw JSON: [`scheduler-concurrency.json`](scheduler-concurrency.json). + +Speedup vs. `concurrency=1, platform` baseline: + +| concurrency | platform speedup | virtual speedup | +|---|---|---| +| 4 | **3.6×** | 3.4× | +| 16 | **6.2×** | 6.2× | +| 64 | **6.6×** | 6.2× | + +## The ticket's hypothesis did not hold — reporting it straight + +The ticket's baseline expectations were: +- `concurrency=4, platform`: target ~3-4× single-threaded — **confirmed** (3.6×). +- `concurrency=16, virtual` should outperform `concurrency=16, platform` by 15-25% — **not + observed**. They're tied (0% difference, well within error bars). +- `concurrency=64+, virtual` should be "the only viable option" — **not observed**. Platform is + actually ~7% *faster* than virtual at concurrency=64. + +At no tested concurrency level does virtual meaningfully beat platform; platform is same-or-ahead +throughout. Since this run already has JEP 491 (JDK 25), the usual "virtual threads regress +because of pre-JDK-24 pinning" explanation doesn't apply here. The more likely explanation: +virtual threads' advantage shows up when task count vastly exceeds available platform threads +(hundreds to thousands of concurrent blocking tasks contending for a small carrier pool) — this +benchmark only ever runs *exactly* `concurrency` tasks against a platform pool sized to exactly +`concurrency`, so there's no oversubscription for virtual threads to fix, and they pay their own +(small) continuation-management overhead for no return. The ticket's "64+" ceiling likely needs to +be an order of magnitude higher (hundreds to low thousands) before virtual threads' benefit would +show up — genuinely testing that means decoupling worker count from DB connection limits, which +is out of scope here. + +**Practical takeaway: default to platform threads (`OutboxSchedulerConfig`'s default) at every +concurrency level tested (1-64).** `workerExecutorFactory` remains available for users who want to +verify virtual threads at concurrency well beyond 64 for their own workload, but nothing in this +data recommends switching the default. + +## Where the scaling actually plateaus + +Sublinear scaling from `concurrency=4` (3.6×) to `concurrency=16` (6.2×) to `concurrency=64` +(6.6×) — far from the linear 4×/16×/64× the "multiplies throughput linearly" framing in the ticket +might suggest. Two contributing factors: +- **Shared, single-instance backend.** One Postgres container and one Kafka broker mean workers + increasingly contend for the same connection acceptance path and broker I/O as concurrency + rises — this benchmark measures the fan-out mechanism's overhead, not unlimited horizontal DB + scaling, and the "concurrency × instances ≤ `max_connections` / 2" rule of thumb below exists + precisely because of this. +- **Round-count artifact at high concurrency.** With `batchSize=100` fixed, `concurrency=64` claims + all 6,400 entries in a *single* round — so its measurement is dominated by one-time per-round + costs (opening `concurrency` fresh connections, `claimPending` query planning) with no further + rounds to amortize them over. Lower concurrency values run many rounds, averaging out that fixed + cost. This means the `concurrency=64` number is a slight pessimistic outlier relative to a + steady-state, many-round production workload — a caveat worth keeping in mind rather than reading + 64 as a hard ceiling. + +## Tuning recommendation + +- **`concurrency=4` to `concurrency=16`** is the practical sweet spot for a single Postgres+Kafka + backend on this hardware: most of the available speedup (3.6×-6.2×) is already captured, and + gains beyond 16 are marginal (6.2× → 6.6×) while consuming more DB connections. +- Keep the default `defaultPlatformPool` executor. Nothing in this data supports switching to + `virtualThreadPool` at the concurrency levels tested. +- **Rule of thumb documented on `OutboxSchedulerConfig`:** `concurrency × instances ≤ + max_connections / 2` — cross-instance coordination is free (`FOR UPDATE SKIP LOCKED` handles + disjoint claims across both workers and instances with no app-level coordination), but every + worker across every instance holds a DB connection for the duration of its batch. + +## Verification context + +- Unit tests: `OutboxSchedulerConfigTest` (concurrency validation 1-256, `defaultPlatformPool` + thread naming, `virtualThreadPool` runs on virtual threads), `OutboxSchedulerTest` (fan-out + dispatches exactly `concurrency` workers per tick, `concurrency=1` never touches + `workerExecutorFactory`, one worker's exception doesn't block the others). +- Integration test: `ConcurrentClaimTests` extended with a real `OutboxScheduler` + (`concurrency=4`) against both Postgres and MySQL — disjoint claims across workers *and* ticks, + zero delivery amplification. +- `okapi-spring-boot`: `okapi.processor.concurrency` property wired through, with startup-failure + coverage for invalid values. +- ktlint clean, full `./gradlew build` green (core, spring-boot, integration-tests). diff --git a/benchmarks/scheduler-concurrency.json b/benchmarks/scheduler-concurrency.json new file mode 100644 index 0000000..d29d8d0 --- /dev/null +++ b/benchmarks/scheduler-concurrency.json @@ -0,0 +1,532 @@ +[ + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "1", + "executorType" : "platform" + }, + "primaryMetric" : { + "score" : 0.3566704235731962, + "scoreError" : 0.0974459790021155, + "scoreConfidence" : [ + 0.25922444457108074, + 0.4541164025753117 + ], + "scorePercentiles" : { + "0.0" : 0.26312747039772727, + "50.0" : 0.3689720704123264, + "90.0" : 0.45276361384374997, + "95.0" : 0.45603878162946426, + "99.0" : 0.45603878162946426, + "99.9" : 0.45603878162946426, + "99.99" : 0.45603878162946426, + "99.999" : 0.45603878162946426, + "99.9999" : 0.45603878162946426, + "100.0" : 0.45603878162946426 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.45603878162946426, + 0.356075834140625, + 0.3919755042013889, + 0.4232871037723214, + 0.3602697251215278 + ], + [ + 0.37839303164930554, + 0.377674415703125, + 0.29365464715625, + 0.26312747039772727, + 0.2662077219602273 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "1", + "executorType" : "virtual" + }, + "primaryMetric" : { + "score" : 0.34720389351516, + "scoreError" : 0.06952627072990189, + "scoreConfidence" : [ + 0.2776776227852581, + 0.41673016424506193 + ], + "scorePercentiles" : { + "0.0" : 0.30811317578125, + "50.0" : 0.33832400247395833, + "90.0" : 0.4573723407770648, + "95.0" : 0.4672377064732143, + "99.0" : 0.4672377064732143, + "99.9" : 0.4672377064732143, + "99.99" : 0.4672377064732143, + "99.999" : 0.4672377064732143, + "99.9999" : 0.4672377064732143, + "100.0" : 0.4672377064732143 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.33815907192708333, + 0.36858404951171875, + 0.3288113462152778, + 0.3421425101215278, + 0.4672377064732143 + ], + [ + 0.3099344863125, + 0.30811317578125, + 0.3203615924375, + 0.35020606335069443, + 0.3384889330208333 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "4", + "executorType" : "platform" + }, + "primaryMetric" : { + "score" : 0.09862723203400736, + "scoreError" : 0.002815739354474301, + "scoreConfidence" : [ + 0.09581149267953307, + 0.10144297138848166 + ], + "scorePercentiles" : { + "0.0" : 0.09610727442095589, + "50.0" : 0.09840320409007353, + "90.0" : 0.1019284712601103, + "95.0" : 0.10213047217830883, + "99.0" : 0.10213047217830883, + "99.9" : 0.10213047217830883, + "99.99" : 0.10213047217830883, + "99.999" : 0.10213047217830883, + "99.9999" : 0.10213047217830883, + "100.0" : 0.10213047217830883 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.09976032856617648, + 0.09823047260110294, + 0.09805603017463235, + 0.09751775696691177, + 0.09611523820772058 + ], + [ + 0.10213047217830883, + 0.10011046299632353, + 0.09966834864889706, + 0.09857593557904412, + 0.09610727442095589 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "4", + "executorType" : "virtual" + }, + "primaryMetric" : { + "score" : 0.10397616305457262, + "scoreError" : 0.002796758328300158, + "scoreConfidence" : [ + 0.10117940472627246, + 0.10677292138287278 + ], + "scorePercentiles" : { + "0.0" : 0.10130346124080883, + "50.0" : 0.10362827512408088, + "90.0" : 0.10692789718307676, + "95.0" : 0.10698128563419118, + "99.0" : 0.10698128563419118, + "99.9" : 0.10698128563419118, + "99.99" : 0.10698128563419118, + "99.999" : 0.10698128563419118, + "99.9999" : 0.10698128563419118, + "100.0" : 0.10698128563419118 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.10644740112304688, + 0.10289676816176471, + 0.10356370519301471, + 0.10539989581054687, + 0.10177409924632352 + ], + [ + 0.10698128563419118, + 0.10353915592830883, + 0.10369284505514706, + 0.10416301315257354, + 0.10130346124080883 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "16", + "executorType" : "platform" + }, + "primaryMetric" : { + "score" : 0.05831040687920025, + "scoreError" : 0.005466810140024821, + "scoreConfidence" : [ + 0.05284359673917543, + 0.06377721701922506 + ], + "scorePercentiles" : { + "0.0" : 0.0537906276328125, + "50.0" : 0.057420186859786185, + "90.0" : 0.06411888988465073, + "95.0" : 0.06418689417534722, + "99.0" : 0.06418689417534722, + "99.9" : 0.06418689417534722, + "99.99" : 0.06418689417534722, + "99.999" : 0.06418689417534722, + "99.9999" : 0.06418689417534722, + "100.0" : 0.06418689417534722 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.0554868720703125, + 0.055249909484375, + 0.0554496171640625, + 0.05699982454769737, + 0.0537906276328125 + ], + [ + 0.06072154741776316, + 0.06350685126838235, + 0.06418689417534722, + 0.057840549171875, + 0.059871375859375 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "16", + "executorType" : "virtual" + }, + "primaryMetric" : { + "score" : 0.05829309122861842, + "scoreError" : 0.0020862941345980135, + "scoreConfidence" : [ + 0.056206797094020405, + 0.06037938536321644 + ], + "scorePercentiles" : { + "0.0" : 0.0567883317109375, + "50.0" : 0.057635347, + "90.0" : 0.061058549313322365, + "95.0" : 0.0611839275, + "99.0" : 0.0611839275, + "99.9" : 0.0611839275, + "99.99" : 0.0611839275, + "99.999" : 0.0611839275, + "99.9999" : 0.0611839275, + "100.0" : 0.0611839275 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.057609984046875, + 0.0611839275, + 0.05911475393914474, + 0.0574009990078125, + 0.0567883317109375 + ], + [ + 0.059930145633223686, + 0.0583794749453125, + 0.057660709953125, + 0.0573724303359375, + 0.05749015521381579 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "64", + "executorType" : "platform" + }, + "primaryMetric" : { + "score" : 0.05417698885355184, + "scoreError" : 0.002214712391078403, + "scoreConfidence" : [ + 0.051962276462473436, + 0.05639170124463025 + ], + "scorePercentiles" : { + "0.0" : 0.05227146646875, + "50.0" : 0.053629043937499996, + "90.0" : 0.05614647789638158, + "95.0" : 0.056150869309210524, + "99.0" : 0.056150869309210524, + "99.9" : 0.056150869309210524, + "99.99" : 0.056150869309210524, + "99.999" : 0.056150869309210524, + "99.9999" : 0.056150869309210524, + "100.0" : 0.056150869309210524 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.054760251625, + 0.05347714159375, + 0.053248665359375, + 0.05227146646875, + 0.0526747985234375 + ], + [ + 0.05378094628125, + 0.0559741119765625, + 0.05332468221726191, + 0.056106955180921056, + 0.056150869309210524 + ] + ] + }, + "secondaryMetrics" : { + } + }, + { + "jmhVersion" : "1.37", + "benchmark" : "com.softwaremill.okapi.benchmarks.OutboxSchedulerConcurrencyBenchmark.drainAll", + "mode" : "avgt", + "threads" : 1, + "forks" : 2, + "jvm" : "25.0.2-open", + "jvmArgs" : [ + "-Xms8g", + "-Xmx8g", + "-XX:+UseG1GC", + "-Dliquibase.duplicateFileMode=WARN" + ], + "jdkVersion" : "25.0.2", + "vmName" : "OpenJDK 64-Bit Server VM", + "vmVersion" : "25.0.2+10-69", + "warmupIterations" : 3, + "warmupTime" : "10 s", + "warmupBatchSize" : 1, + "measurementIterations" : 5, + "measurementTime" : "30 s", + "measurementBatchSize" : 1, + "params" : { + "concurrency" : "64", + "executorType" : "virtual" + }, + "primaryMetric" : { + "score" : 0.05779675429407895, + "scoreError" : 0.0025192637530646158, + "scoreConfidence" : [ + 0.05527749054101433, + 0.060316018047143566 + ], + "scorePercentiles" : { + "0.0" : 0.0557219973984375, + "50.0" : 0.05757087195867598, + "90.0" : 0.06081344351475694, + "95.0" : 0.060930749071180554, + "99.0" : 0.060930749071180554, + "99.9" : 0.060930749071180554, + "99.99" : 0.060930749071180554, + "99.999" : 0.060930749071180554, + "99.9999" : 0.060930749071180554, + "100.0" : 0.060930749071180554 + }, + "scoreUnit" : "ms/op", + "rawData" : [ + [ + 0.059757693506944445, + 0.056381873390625, + 0.056691587875, + 0.0557219973984375, + 0.056412676765625 + ], + [ + 0.059110762359375, + 0.060930749071180554, + 0.0577293069765625, + 0.05781845865625, + 0.05741243694078947 + ] + ] + }, + "secondaryMetrics" : { + } + } +] + + diff --git a/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt new file mode 100644 index 0000000..d01d303 --- /dev/null +++ b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/OutboxSchedulerConcurrencyBenchmark.kt @@ -0,0 +1,121 @@ +package com.softwaremill.okapi.benchmarks + +import com.softwaremill.okapi.benchmarks.support.KafkaBenchmarkSupport +import com.softwaremill.okapi.benchmarks.support.PostgresBenchmarkSupport +import com.softwaremill.okapi.core.OutboxEntryProcessor +import com.softwaremill.okapi.core.OutboxMessage +import com.softwaremill.okapi.core.OutboxProcessor +import com.softwaremill.okapi.core.OutboxPublisher +import com.softwaremill.okapi.core.OutboxSchedulerConfig +import com.softwaremill.okapi.core.RetryPolicy +import com.softwaremill.okapi.kafka.KafkaDeliveryInfo +import com.softwaremill.okapi.kafka.KafkaMessageDeliverer +import com.softwaremill.okapi.postgres.PostgresOutboxStore +import org.apache.kafka.clients.producer.KafkaProducer +import org.openjdk.jmh.annotations.Benchmark +import org.openjdk.jmh.annotations.BenchmarkMode +import org.openjdk.jmh.annotations.Level +import org.openjdk.jmh.annotations.Mode +import org.openjdk.jmh.annotations.OperationsPerInvocation +import org.openjdk.jmh.annotations.OutputTimeUnit +import org.openjdk.jmh.annotations.Param +import org.openjdk.jmh.annotations.Scope +import org.openjdk.jmh.annotations.Setup +import org.openjdk.jmh.annotations.State +import org.openjdk.jmh.annotations.TearDown +import java.time.Clock +import java.util.UUID +import java.util.concurrent.Callable +import java.util.concurrent.ExecutorService +import java.util.concurrent.TimeUnit + +/** + * Measures scheduler fan-out throughput (KOJAK-77): each round dispatches [concurrency] + * parallel [OutboxProcessor.processNext] calls -- one per worker, each on its own JDBC + * transaction/connection, exactly mirroring what [com.softwaremill.okapi.core.OutboxScheduler]'s + * internal fan-out does -- and finds the platform-vs-virtual-thread breakeven point. + * + * The scheduler's [java.util.concurrent.ScheduledExecutorService] polling loop is bypassed + * (same rationale as [KafkaThroughputBenchmark]): we measure worker-pool throughput, not + * polling cadence. Kafka `deliverBatch` is the transport, since it's the one delivery path + * fast enough that the fan-out itself -- not delivery I/O -- is the bottleneck being measured + * (HTTP `deliverBatch` isn't implemented yet; KOJAK-74). + */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +open class OutboxSchedulerConcurrencyBenchmark { + + @Param("platform", "virtual") + var executorType: String = "platform" + + @Param("1", "4", "16", "64") + var concurrency: Int = 1 + + private lateinit var postgres: PostgresBenchmarkSupport + private lateinit var kafka: KafkaBenchmarkSupport + private lateinit var producer: KafkaProducer + private lateinit var publisher: OutboxPublisher + private lateinit var processor: OutboxProcessor + private lateinit var executor: ExecutorService + private lateinit var topic: String + + @Setup(Level.Trial) + fun setupTrial() { + postgres = PostgresBenchmarkSupport().also { it.start() } + kafka = KafkaBenchmarkSupport().also { it.start() } + producer = kafka.createProducer() + topic = "bench-${UUID.randomUUID()}" + + val clock = Clock.systemUTC() + val store = PostgresOutboxStore(postgres.jdbc) + publisher = OutboxPublisher(store, clock) + val deliverer = KafkaMessageDeliverer(producer) + val entryProcessor = OutboxEntryProcessor(deliverer, RetryPolicy(maxRetries = 0), clock) + processor = OutboxProcessor(store, entryProcessor) + + executor = when (executorType) { + "virtual" -> OutboxSchedulerConfig.virtualThreadPool(concurrency) + else -> OutboxSchedulerConfig.defaultPlatformPool(concurrency) + } + } + + @Setup(Level.Invocation) + fun setupInvocation() { + postgres.truncate() + val info = KafkaDeliveryInfo(topic = topic, partitionKey = "k") + postgres.jdbc.withTransaction { + repeat(TOTAL_ENTRIES) { + publisher.publish(OutboxMessage("bench.event", PAYLOAD), info) + } + } + } + + @Benchmark + @OperationsPerInvocation(TOTAL_ENTRIES) + fun drainAll() { + var remaining = TOTAL_ENTRIES + while (remaining > 0) { + val futures = (1..concurrency).map { + executor.submit(Callable { postgres.jdbc.withTransaction { processor.processNext(BATCH_SIZE) } }) + } + val processedThisRound = futures.sumOf { it.get() } + if (processedThisRound == 0) break + remaining -= processedThisRound + } + } + + @TearDown(Level.Trial) + fun teardown() { + executor.shutdown() + producer.close() + kafka.stop() + postgres.stop() + } + + companion object { + const val TOTAL_ENTRIES = 6400 + const val BATCH_SIZE = 100 + private const val PAYLOAD = """{"orderId":"order-42","amount":100.50,"currency":"EUR"}""" + } +} diff --git a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt index f2bbb2e..3bc17b2 100644 --- a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt +++ b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxScheduler.kt @@ -1,6 +1,7 @@ package com.softwaremill.okapi.core import org.slf4j.LoggerFactory +import java.util.concurrent.ExecutorService import java.util.concurrent.Executors import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.TimeUnit @@ -9,9 +10,16 @@ import java.util.concurrent.atomic.AtomicBoolean /** * Standalone scheduler that periodically calls [OutboxProcessor.processNext]. * - * Each tick runs inside [transactionRunner]. The runner is required: without a surrounding - * transaction, `FOR UPDATE SKIP LOCKED` releases its row lock at the end of the claim - * SELECT and concurrent processor instances deliver the same entry multiple times. + * Each worker's call runs inside [transactionRunner]. The runner is required: without a + * surrounding transaction, `FOR UPDATE SKIP LOCKED` releases its row lock at the end of the + * claim SELECT and concurrent processor instances deliver the same entry multiple times. + * + * With [OutboxSchedulerConfig.concurrency] `> 1`, each tick fans out to that many workers on + * [OutboxSchedulerConfig.workerExecutorFactory]'s pool, each independently claiming and + * processing its own batch -- `FOR UPDATE SKIP LOCKED` guarantees disjoint claims, so no + * app-level coordination is needed. The tick waits for every worker before the next scheduled + * interval, so ticks never overlap. `concurrency = 1` (default) skips the executor entirely + * and calls the batch inline, preserving the original zero-overhead single-worker behavior. * * Runs on a single daemon thread with explicit [start]/[stop] lifecycle. * [start] and [stop] are single-use -- the internal executor cannot be restarted after shutdown. @@ -33,10 +41,18 @@ class OutboxScheduler @JvmOverloads constructor( Thread(r, "outbox-processor").apply { isDaemon = true } } + private val workers: ExecutorService? = + if (config.concurrency > 1) config.workerExecutorFactory(config.concurrency) else null + fun start() { check(!scheduler.isShutdown) { "OutboxScheduler cannot be restarted after stop()" } if (!running.compareAndSet(false, true)) return - logger.info("Outbox processor started [interval={}, batchSize={}]", config.interval, config.batchSize) + logger.info( + "Outbox processor started [interval={}, batchSize={}, concurrency={}]", + config.interval, + config.batchSize, + config.concurrency, + ) scheduler.scheduleWithFixedDelay(::tick, 0L, config.interval.toMillis(), TimeUnit.MILLISECONDS) } @@ -46,12 +62,26 @@ class OutboxScheduler @JvmOverloads constructor( if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) { scheduler.shutdownNow() } + workers?.let { + it.shutdown() + if (!it.awaitTermination(5, TimeUnit.SECONDS)) it.shutdownNow() + } logger.info("Outbox processor stopped") } fun isRunning(): Boolean = running.get() private fun tick() { + val pool = workers + if (pool == null) { + processBatch() + } else { + val futures = (1..config.concurrency).map { pool.submit(::processBatch) } + futures.forEach { it.get() } + } + } + + private fun processBatch() { try { transactionRunner.runInTransaction { outboxProcessor.processNext(config.batchSize) } logger.debug("Outbox processor tick completed [batchSize={}]", config.batchSize) diff --git a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt index 5b6c41f..4df9138 100644 --- a/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt +++ b/okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfig.kt @@ -1,13 +1,49 @@ package com.softwaremill.okapi.core import java.time.Duration +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicInteger data class OutboxSchedulerConfig( val interval: Duration = Duration.ofSeconds(1), val batchSize: Int = 10, + /** + * Number of parallel workers fanned out per scheduler tick. Each worker claims its own + * batch via [OutboxStore.claimPending]'s `FOR UPDATE SKIP LOCKED`, so workers never + * compete for the same rows. `1` (default) preserves the original single-worker tick + * with zero pooling overhead. + */ + val concurrency: Int = 1, + /** + * Builds the [ExecutorService] used to run workers when [concurrency] > 1. Defaults to + * a fixed platform-thread pool ([defaultPlatformPool]); pass [virtualThreadPool] for + * high concurrency (roughly 32+) to avoid platform-thread context-switch overhead. + */ + val workerExecutorFactory: (Int) -> ExecutorService = ::defaultPlatformPool, ) { init { require(!interval.isNegative && interval.toMillis() > 0) { "interval must be at least 1ms, got: $interval" } require(batchSize > 0) { "batchSize must be positive, got: $batchSize" } + require(concurrency in 1..256) { "concurrency must be between 1 and 256, got: $concurrency" } + } + + companion object { + /** Fixed pool of [n] daemon platform threads, named `outbox-worker-N`. */ + @JvmStatic + fun defaultPlatformPool(n: Int): ExecutorService { + val counter = AtomicInteger(0) + return Executors.newFixedThreadPool(n) { r -> + Thread(r, "outbox-worker-${counter.incrementAndGet()}").apply { isDaemon = true } + } + } + + /** + * One virtual thread per submitted task. [n] is accepted for signature parity with + * [defaultPlatformPool] but unused — virtual thread creation is unbounded. + */ + @JvmStatic + @Suppress("UNUSED_PARAMETER") + fun virtualThreadPool(n: Int): ExecutorService = Executors.newVirtualThreadPerTaskExecutor() } } diff --git a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt index da30e6d..165513e 100644 --- a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt +++ b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerConfigTest.kt @@ -7,6 +7,8 @@ import java.time.Duration import java.time.Duration.ofMillis import java.time.Duration.ofNanos import java.time.Duration.ofSeconds +import java.util.concurrent.Callable +import java.util.concurrent.TimeUnit class OutboxSchedulerConfigTest : FunSpec({ @@ -14,12 +16,14 @@ class OutboxSchedulerConfigTest : FunSpec({ val config = OutboxSchedulerConfig() config.interval shouldBe ofSeconds(1) config.batchSize shouldBe 10 + config.concurrency shouldBe 1 } test("accepts custom valid values") { - val config = OutboxSchedulerConfig(interval = ofMillis(500), batchSize = 50) + val config = OutboxSchedulerConfig(interval = ofMillis(500), batchSize = 50, concurrency = 8) config.interval shouldBe ofMillis(500) config.batchSize shouldBe 50 + config.concurrency shouldBe 8 } test("rejects zero interval") { @@ -51,4 +55,48 @@ class OutboxSchedulerConfigTest : FunSpec({ OutboxSchedulerConfig(interval = ofNanos(1)) } } + + test("accepts concurrency boundaries 1 and 256") { + OutboxSchedulerConfig(concurrency = 1).concurrency shouldBe 1 + OutboxSchedulerConfig(concurrency = 256).concurrency shouldBe 256 + } + + test("rejects zero concurrency") { + shouldThrow { + OutboxSchedulerConfig(concurrency = 0) + } + } + + test("rejects negative concurrency") { + shouldThrow { + OutboxSchedulerConfig(concurrency = -1) + } + } + + test("rejects concurrency above 256") { + shouldThrow { + OutboxSchedulerConfig(concurrency = 257) + } + } + + test("defaultPlatformPool runs submitted tasks and names its threads outbox-worker-N") { + val pool = OutboxSchedulerConfig.defaultPlatformPool(2) + try { + val threadNames = (1..2).map { pool.submit(Callable { Thread.currentThread().name }) } + .map { it.get(2, TimeUnit.SECONDS) } + threadNames.toSet() shouldBe setOf("outbox-worker-1", "outbox-worker-2") + } finally { + pool.shutdown() + } + } + + test("virtualThreadPool runs submitted tasks on virtual threads") { + val pool = OutboxSchedulerConfig.virtualThreadPool(4) + try { + val isVirtual = pool.submit(Callable { Thread.currentThread().isVirtual }).get(2, TimeUnit.SECONDS) + isVirtual shouldBe true + } finally { + pool.shutdown() + } + } }) diff --git a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt index 98af3ee..2c7cbd2 100644 --- a/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt +++ b/okapi-core/src/test/kotlin/com/softwaremill/okapi/core/OutboxSchedulerTest.kt @@ -131,6 +131,77 @@ class OutboxSchedulerTest : FunSpec({ txInvoked.get() shouldBe true } + + test("concurrency > 1 fans out to that many workers per tick") { + val callCount = AtomicInteger(0) + val threadNames = java.util.Collections.synchronizedSet(mutableSetOf()) + val latch = CountDownLatch(4) + val processor = stubProcessor { _ -> + callCount.incrementAndGet() + threadNames += Thread.currentThread().name + latch.countDown() + } + + val scheduler = OutboxScheduler( + outboxProcessor = processor, + transactionRunner = noOpTransactionRunner(), + config = OutboxSchedulerConfig(interval = ofMinutes(1), concurrency = 4), + ) + + scheduler.start() + latch.await(2, TimeUnit.SECONDS) shouldBe true + scheduler.stop() + + callCount.get() shouldBe 4 + threadNames shouldBe setOf("outbox-worker-1", "outbox-worker-2", "outbox-worker-3", "outbox-worker-4") + } + + test("concurrency = 1 never invokes workerExecutorFactory (zero-overhead default path)") { + val factoryCalled = AtomicBoolean(false) + val latch = CountDownLatch(1) + val processor = stubProcessor { _ -> latch.countDown() } + + val scheduler = OutboxScheduler( + outboxProcessor = processor, + transactionRunner = noOpTransactionRunner(), + config = OutboxSchedulerConfig( + interval = ofMillis(50), + concurrency = 1, + workerExecutorFactory = { n -> + factoryCalled.set(true) + OutboxSchedulerConfig.defaultPlatformPool(n) + }, + ), + ) + + scheduler.start() + latch.await(2, TimeUnit.SECONDS) shouldBe true + scheduler.stop() + + factoryCalled.get() shouldBe false + } + + test("one worker's exception does not prevent the other workers from completing") { + val callCount = AtomicInteger(0) + val latch = CountDownLatch(4) + val processor = stubProcessor { _ -> + val count = callCount.incrementAndGet() + latch.countDown() + if (count == 1) throw RuntimeException("worker failure") + } + + val scheduler = OutboxScheduler( + outboxProcessor = processor, + transactionRunner = noOpTransactionRunner(), + config = OutboxSchedulerConfig(interval = ofMinutes(1), concurrency = 4), + ) + + scheduler.start() + latch.await(2, TimeUnit.SECONDS) shouldBe true + scheduler.stop() + + callCount.get() shouldBe 4 + } }) private fun stubProcessor(onProcessNext: (Int) -> Unit): OutboxProcessor { diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt index 55f1fd6..73142d2 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/concurrency/ConcurrentClaimTests.kt @@ -6,9 +6,12 @@ import com.softwaremill.okapi.core.OutboxEntryProcessor import com.softwaremill.okapi.core.OutboxId import com.softwaremill.okapi.core.OutboxMessage import com.softwaremill.okapi.core.OutboxProcessor +import com.softwaremill.okapi.core.OutboxScheduler +import com.softwaremill.okapi.core.OutboxSchedulerConfig import com.softwaremill.okapi.core.OutboxStatus import com.softwaremill.okapi.core.OutboxStore import com.softwaremill.okapi.core.RetryPolicy +import com.softwaremill.okapi.core.TransactionRunner import com.softwaremill.okapi.test.support.JdbcConnectionProvider import com.softwaremill.okapi.test.support.RecordingMessageDeliverer import io.kotest.assertions.withClue @@ -18,6 +21,7 @@ import io.kotest.matchers.maps.shouldContain import io.kotest.matchers.shouldBe import java.sql.Connection import java.time.Clock +import java.time.Duration import java.time.Instant import java.time.ZoneOffset import java.util.concurrent.CompletableFuture @@ -165,4 +169,49 @@ fun FunSpec.concurrentClaimTests( counts shouldContain (OutboxStatus.PENDING to 0L) } } + + test("[$dbName] OutboxScheduler with concurrency=4 workers processes disjoint batches without amplification") { + val fixedClock = Clock.fixed(Instant.parse("2024-01-01T00:00:00Z"), ZoneOffset.UTC) + + // 80 entries, batchSize=10, concurrency=4 -> 40 claimed per tick, so it takes 2 ticks + // to drain everything. This exercises both the multi-worker fan-out within a single tick + // AND repeated ticks, unlike the single-shot manual-thread test above. + val entryCount = 80 + jdbc.withTransaction { + (0 until entryCount).forEach { i -> store.persist(createTestEntry(i)) } + } + + val recorder = RecordingMessageDeliverer() + val entryProcessor = OutboxEntryProcessor(recorder, RetryPolicy(maxRetries = 0), fixedClock) + val processor = OutboxProcessor(store, entryProcessor) + + val transactionRunner = object : TransactionRunner { + override fun runInTransaction(block: () -> T): T = + jdbc.withTransaction(transactionIsolation = Connection.TRANSACTION_READ_COMMITTED) { block() } + } + + val scheduler = OutboxScheduler( + outboxProcessor = processor, + transactionRunner = transactionRunner, + config = OutboxSchedulerConfig(interval = Duration.ofMillis(50), batchSize = 10, concurrency = 4), + ) + + scheduler.start() + val deadline = System.currentTimeMillis() + 30_000 + while (recorder.deliveryCount() < entryCount && System.currentTimeMillis() < deadline) { + Thread.sleep(50) + } + scheduler.stop() + + recorder.assertNoAmplification() + withClue("Expected exactly $entryCount unique deliveries from a concurrency=4 scheduler, got ${recorder.deliveryCount()}") { + recorder.deliveryCount() shouldBe entryCount + } + + val counts = jdbc.withTransaction { store.countByStatuses() } + withClue("DB state after scheduler run: $counts") { + counts shouldContain (OutboxStatus.DELIVERED to entryCount.toLong()) + counts shouldContain (OutboxStatus.PENDING to 0L) + } + } } diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt index e575aa6..9d9dfc9 100644 --- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt +++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt @@ -205,6 +205,7 @@ class OutboxAutoConfiguration( config = OutboxSchedulerConfig( interval = props.interval, batchSize = props.batchSize, + concurrency = props.concurrency, ), ) } diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt index 5c1f077..f7a8fb3 100644 --- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt +++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorProperties.kt @@ -8,10 +8,16 @@ data class OutboxProcessorProperties( val interval: Duration = Duration.ofSeconds(1), val batchSize: Int = 10, val maxRetries: Int = 5, + /** + * Number of parallel workers fanned out per scheduler tick, each claiming its own batch + * via `FOR UPDATE SKIP LOCKED`. See [com.softwaremill.okapi.core.OutboxSchedulerConfig]. + */ + val concurrency: Int = 1, ) { init { require(!interval.isNegative && interval.toMillis() > 0) { "interval must be at least 1ms" } require(batchSize > 0) { "batchSize must be positive" } require(maxRetries >= 0) { "maxRetries must be >= 0" } + require(concurrency in 1..256) { "concurrency must be between 1 and 256" } } } diff --git a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt index 56c7f86..fcd5239 100644 --- a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt +++ b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxProcessorAutoConfigurationTest.kt @@ -50,12 +50,14 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({ "okapi.processor.interval=500ms", "okapi.processor.batch-size=20", "okapi.processor.max-retries=3", + "okapi.processor.concurrency=4", ) .run { ctx -> val props = ctx.getBean(OutboxProcessorProperties::class.java) props.interval shouldBe ofMillis(500) props.batchSize shouldBe 20 props.maxRetries shouldBe 3 + props.concurrency shouldBe 4 } } @@ -65,6 +67,7 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({ props.interval shouldBe ofSeconds(1) props.batchSize shouldBe 10 props.maxRetries shouldBe 5 + props.concurrency shouldBe 1 } } @@ -92,6 +95,14 @@ class OutboxProcessorAutoConfigurationTest : FunSpec({ } } + test("invalid concurrency triggers startup failure") { + contextRunner + .withPropertyValues("okapi.processor.concurrency=0") + .run { ctx -> + ctx.startupFailure.shouldNotBeNull() + } + } + test("stop(callback) invokes callback AND actually halts the scheduler") { contextRunner.run { ctx -> val scheduler = ctx.getBean(OutboxProcessorScheduler::class.java)