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?