feat: Kafka deliverBatch fire-flush-await — 13-41x throughput (KOJAK-73)

[endrju19]

Summary

Overrides KafkaMessageDeliverer.deliverBatch with the fire-flush-await pattern. Replaces N sequential producer.send().get() round-trips with one batched network round-trip via producer.flush().

KOJAK-73 — Kafka deliverBatch
Builds on KOJAK-72 (deliverBatch interface).

Headline numbers

Smoke benchmark vs KOJAK-68 baseline (same hardware: MacBook M3 Max, JDK 25 LTS, Postgres 16 + Kafka 3.8.1 via Testcontainers, fork=1, warmup=1, iter=2, warmup=10s, measurement=15s):

batchSize	Baseline	This PR	Improvement
10	~109 msg/s	~1,468 msg/s	13.5×
50	~115 msg/s	~3,731 msg/s	32.3×
100	~115 msg/s	~4,717 msg/s	41.0×

batchSize is now load-bearing. Pre-KOJAK-73 throughput was flat across batchSize values — the bottleneck was per-record blocking producer.send().get(). Post-KOJAK-73 throughput scales with batchSize, proving Kafka's internal record batching is being exploited.

Full results + interpretation: benchmarks/results-postopt-kojak-73.md.

Implementation

override fun deliverBatch(entries: List<OutboxEntry>): List<Pair<OutboxEntry, DeliveryResult>> {
    if (entries.isEmpty()) return emptyList()

    // Phase 1: fire all (non-blocking, records queue in producer buffer)
    val inflight = entries.map { it to fireOne(it) }

    // Phase 2: flush — bypasses linger.ms, one batched RTT for all in-flight
    try { producer.flush() } catch (e: InterruptException) {
        Thread.currentThread().interrupt()
    }

    // Phase 3: collect — Future.get() is non-blocking after flush
    return inflight.map { (entry, outcome) -> entry to awaitOne(outcome) }
}

SendOutcome sealed type distinguishes synchronous send exceptions (e.g., BufferExhaustedException, SerializationException) from in-flight futures — one failing send never aborts the batch.

deliver() refactored to share buildRecord / classifyException helpers with deliverBatch — no behavior change for single-entry path.

Test coverage

New unit tests in KafkaMessageDelivererBatchTest:

• Empty input → empty output, producer untouched
• All-success preserves input order
• Single flush() call (verified via flush counter override)
• Synchronous send exception → per-entry classification (Permanent + Retriable variants)
• Future-based async exception (via MockProducer override that drives completeNext/errorNext from inside flush)

Integration tests in okapi-integration-tests continue to pass with real Kafka.

Sublinear scaling: KOJAK-75 motivation

Going from batchSize=50 → 100 gives 32× → 41× (only ~26% more, not 2×). At larger batches, the N individual updateAfterProcessing calls become significant relative to the now-fast Kafka path. This is exactly what KOJAK-75 (batch UPDATE via executeBatch) addresses — it'll be the next big lever once HTTP deliverBatch (KOJAK-74) lands.

Test plan

• ./gradlew test ktlintCheck -x :okapi-integration-tests:test — all unit tests + ktlint pass
• ./gradlew :okapi-integration-tests:test — full Testcontainers suite (Postgres + MySQL + Kafka + WireMock) passes
• ./gradlew :okapi-benchmarks:jmhJar + smoke Kafka throughput run — captured numbers above

[ramafasa]

maybe better get with timeout?

[endrju19]

bare get() is safe here. Kafka's Sender already bounds settlement via delivery.timeout.ms (default 120 s), and flush() only returns once every Future is settled. Users tune it via the producer config they inject