From 698e2b3b68f1ccb6fd5a8ee0f206f55c0b9a6e89 Mon Sep 17 00:00:00 2001 From: dirkjink Date: Mon, 7 Sep 2026 20:48:30 +0200 Subject: [PATCH 1/2] Fix the findings of the 2026-09-07T20-23 architecture review - M: one BoundedWorkerDispatcher skeleton (queue, worker, in-flight ownership, death handler, two-phase close, reentry mark) beneath SendDispatcher and FallbackDispatcher, which keep only their delivery and their rejection accounting; one ParallelClose helper replaces the two hand-rolled deadline-join loops - M: the configuration guide's defaults quick reference is the canonical statement of the "code" defaults, enforced by DocumentationContractTest against the constants; README links instead of restating the breaker numbers - M: restart after stop() is refused (ADR-0004) - the restart symmetry added on 2026-09-07 (fallback restart, breaker reset, guard re-arm) is removed; the binding keeps deciding on the appender's bound state - L: shared test support (RecordingAppender, RecordingProducerFactory, three encoders) replaces four private recorders, four private producer factories and three private encoders - L: CI compiles the benchmark module against the freshly built library so the regression instrument cannot rot silently - L: guide section "What is deliberately not configurable" with the reason per item, linked from the README extension points Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015GfdGp7eUrjJKBvcJUx3q2 --- .github/workflows/ci.yml | 9 + README.md | 22 +- ...-appender-instances-are-not-restartable.md | 50 +++ docs/config/kafka-appender-config-guide.md | 31 +- .../tabellarium/BoundedWorkerDispatcher.kt | 292 ++++++++++++++++ .../tabellarium/FallbackDispatcher.kt | 315 +++--------------- .../eu/inqudium/tabellarium/KafkaAppender.kt | 114 +++---- .../KafkaAppenderMetricsBinding.kt | 12 +- .../eu/inqudium/tabellarium/ParallelClose.kt | 57 ++++ .../inqudium/tabellarium/ProducerRegistry.kt | 44 +-- .../eu/inqudium/tabellarium/SendDispatcher.kt | 305 ++++------------- .../tabellarium/DocumentationContractTest.kt | 91 +++++ .../tabellarium/FallbackDispatcherTest.kt | 22 +- .../KafkaAppenderMetricsBindingTest.kt | 74 +--- .../inqudium/tabellarium/KafkaAppenderTest.kt | 276 ++++----------- .../tabellarium/KafkaBrokerIntegrationTest.kt | 11 +- .../tabellarium/ProducerRegistryTest.kt | 35 +- .../tabellarium/ResilientMessageSenderTest.kt | 46 +-- .../tabellarium/SendDispatcherTest.kt | 28 +- .../eu/inqudium/tabellarium/TestSupport.kt | 98 ++++++ .../tabellarium/ThreadSafeListAppender.kt | 5 +- 21 files changed, 895 insertions(+), 1042 deletions(-) create mode 100644 docs/adr/ADR-0004-appender-instances-are-not-restartable.md create mode 100644 src/main/kotlin/eu/inqudium/tabellarium/BoundedWorkerDispatcher.kt create mode 100644 src/main/kotlin/eu/inqudium/tabellarium/ParallelClose.kt create mode 100644 src/test/kotlin/eu/inqudium/tabellarium/DocumentationContractTest.kt create mode 100644 src/test/kotlin/eu/inqudium/tabellarium/TestSupport.kt diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ba4eaaa..c599ea3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,6 +37,15 @@ jobs: - name: Build and test run: mvn --batch-mode --no-transfer-progress verify + # The JMH module is a standalone build against the installed + # library snapshot (see benchmarks/README.md). Compiling it here - + # not running it - keeps the regression instrument from rotting + # silently when an internal seam it reaches changes shape. + - name: Compile the benchmark module against the freshly built library + run: | + mvn --batch-mode --no-transfer-progress -DskipTests -Djacoco.skip=true -Dktlint.skip=true -Dcyclonedx.skip=true install + mvn --batch-mode --no-transfer-progress -f benchmarks/pom.xml compile + # Repo-local script instead of a third-party reporting action (this # workflow pins actions to SHAs; a summary needs no new supply-chain # surface). Runs on failure too, so a red build still shows which diff --git a/README.md b/README.md index 1bdf008..81403a1 100644 --- a/README.md +++ b/README.md @@ -287,16 +287,13 @@ Three resilience mechanisms run independently per topic class: 1. **Per-class circuit breaker.** A Resilience4j `CircuitBreaker` is instantiated per active topic class. A stuck audit-topic broker does - not throttle technical-log delivery, and vice versa. Default - thresholds (tuned for logging volume, canonical in the - [configuration guide](docs/config/kafka-appender-config-guide.md)): - 50% failure rate over a sliding window of 20 calls, 30 second - cooldown in open state, 10 probe calls in half-open, probes spread - 5 ms apart. These thresholds are fixed in code - the breaker - registry is an internal seam (ADR-0002), so there is currently no - supported way to tune them per deployment; if a real tuning need - comes up, open an issue so it can become an XML-bindable property - with a follow-up ADR instead of an ad-hoc hook. + not throttle technical-log delivery, and vice versa. The + thresholds are tuned for logging volume and fixed in code; their + canonical values live in the configuration guide's + [defaults quick reference](docs/config/kafka-appender-config-guide.md#12-defaults-quick-reference), + which a test keeps in step with the constants. Why they are not + configurable - and what else is deliberately fixed - is explained in + the guide's [section on fixed behavior](docs/config/kafka-appender-config-guide.md#13-what-is-deliberately-not-configurable). 2. **Asynchronous delivery with callback-driven outcome tracking.** Kafka's `producer.send` is invoked with a callback that feeds the @@ -790,7 +787,10 @@ user id, account id) — most deployments use the trace-id default. Per [ADR-0002](docs/adr/ADR-0002-public-api-is-the-operator-surface.md), such an override would be added as an XML-bindable `KafkaAppender` property (e.g. a partitioning-key MDC name), not by exposing the -internal enricher — open an issue if your deployment needs it. +internal enricher — open an issue if your deployment needs it. The +configuration guide lists +[everything that is deliberately not configurable](docs/config/kafka-appender-config-guide.md#13-what-is-deliberately-not-configurable) +and the reason for each item. ## Future work diff --git a/docs/adr/ADR-0004-appender-instances-are-not-restartable.md b/docs/adr/ADR-0004-appender-instances-are-not-restartable.md new file mode 100644 index 0000000..fcba586 --- /dev/null +++ b/docs/adr/ADR-0004-appender-instances-are-not-restartable.md @@ -0,0 +1,50 @@ +# ADR-0004: Appender instances are not restartable + +- **Status:** accepted +- **Date:** 2026-09-07 +- **Context:** The 2026-09-07 defect analysis + (`docs/assessment/CODE_ANALYSIS-2026-09-07T19-09-00.md`, finding 4) + found that `start()` after `stop()` silently lost the fallback + appender. Its remediation chose the symmetric direction - restart the + fallback, reset the breakers, re-arm the error guard, rebind the + metrics - and the same day's follow-up pass (`.R2.md`, findings R2-4 + and R2-5) and architecture review + (`docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md`, + finding 3) showed what that buys: every per-appender resource now + needs a "what happens on restart?" answer, and no consumer, issue or + README passage asks for a programmatic restart. + +## Decision + +**A `KafkaAppender` instance is started once.** `start()` after +`stop()` is refused with an `addError` naming this ADR; the operator +or program creates a new instance instead. + +Rationale: this is Logback's own lifecycle. A reconfiguration +(``, `LoggerContext.reset()` plus Joran, +Spring Boot's logging-system re-initialization) stops and detaches +every appender and builds **new instances**; `LoggerContext.stop()` is +terminal; Logback's `AsyncAppender` detaches its appenders on `stop()` +and is not restartable in practice. A same-instance restart therefore +only ever originates in application code that holds a reference and +toggles it - a path for which there is no known user. Supporting it +means keeping fallback, circuit-breaker state, metrics binding, +one-shot error report and every future stateful component symmetric +across a second life; refusing it removes those branches and the +interactions between them. + +## Consequences + +- `KafkaAppender.start()` checks the stop guard first and refuses with + a message that names the alternative (a new instance); the guard is + never reset. The Joran round trip and Logback reconfiguration are + unaffected - they never restart an instance. +- The restart-symmetry code introduced on 2026-09-07 (fallback + restart in `start()`, breaker reset in `buildPipeline`, error-guard + re-arm) is removed; the `KafkaAppenderMetricsBinding` keeps deciding + on the appender's bound state, which is simpler than its former + identity set regardless of restart. +- Reversal is a conscious API addition: if a consumer documents a need + for stop/start toggling, a follow-up ADR reinstates the symmetric + lifecycle as a supported contract - with the per-resource answers + written down, not rediscovered. diff --git a/docs/config/kafka-appender-config-guide.md b/docs/config/kafka-appender-config-guide.md index b78ac5c..19e5847 100644 --- a/docs/config/kafka-appender-config-guide.md +++ b/docs/config/kafka-appender-config-guide.md @@ -492,10 +492,9 @@ configuration is tuned for logging traffic: | `waitDurationInOpenState` | `30s` | | `permittedNumberOfCallsInHalfOpenState` | `10` | -These values are fixed in code: the `CircuitBreakerRegistry` is an internal -seam (ADR-0002) and not reachable through the operator surface, so there is -currently no supported per-class override. A demonstrated tuning need should -become an XML-bindable property with a follow-up ADR. +These values are fixed in code (see [section 13](#13-what-is-deliberately-not-configurable) +for why); the table above and the defaults quick reference are kept in step +with the constants by `DocumentationContractTest`. **Ignored exceptions.** The breaker is an *infrastructure-health* signal ("is Kafka reachable?"), not a payload validator. Deterministic, @@ -765,7 +764,29 @@ bound. | Metrics | off (no-op) until bound | code | "code" defaults are not exposed through the XML surface today; they are -listed so operators understand the runtime behavior. +listed so operators understand the runtime behavior. This table is the +**canonical** statement of these numbers: `DocumentationContractTest` +compares every "code" row against the constants in the library, so a +changed constant fails the build until the table follows. The README and +the KDoc link here instead of restating the values. + +--- + +## 13. What is deliberately not configurable + +Everything marked "code" above is fixed on purpose, not by omission. The +operator surface is the public API (ADR-0002); a knob is added when a +demonstrated need exists, as an XML-bindable property with a follow-up ADR +- never as an ad-hoc hook into an internal component. + +| Fixed behavior | Why it is fixed | +| -------------- | --------------- | +| Circuit-breaker thresholds and the half-open probe gap | Tuned for logging traffic (trip after ~10 failures, recover in 30 s, probes spread so a burst cannot burn the half-open window); the registry is an internal seam. No deployment has yet needed different values - open an issue if yours does. | +| Fallback queue capacity and both shutdown drain budgets | Sized so the whole teardown (send drain, producer close, fallback drain) fits a Kubernetes `terminationGracePeriodSeconds: 30` with margin; exposing one budget without the others would let a single setting break that total. | +| `max.block.ms` caps (500 ms; 200 ms for `PERFORMANCE`) | A ceiling that operators may lower but not raise: it bounds each send worker's worst-case stall per event, which is what keeps queue drain during an outage and the shutdown budget predictable. | +| Partitioning-key source (`traceId` in the MDC) | Per ADR-0002 an override would be an XML-bindable appender property; until a deployment asks for one, the trace-id default is the only path (README, "Extension points"). | +| Serializers (`ByteArraySerializer` for key and value) | The appender's wire format; any other serializer would fail every record in the Kafka sender thread. | +| Restart of a stopped appender | Refused by design (ADR-0004): Logback replaces appender instances on reconfiguration instead of restarting them, and the appender follows that lifecycle. | --- diff --git a/src/main/kotlin/eu/inqudium/tabellarium/BoundedWorkerDispatcher.kt b/src/main/kotlin/eu/inqudium/tabellarium/BoundedWorkerDispatcher.kt new file mode 100644 index 0000000..69bc741 --- /dev/null +++ b/src/main/kotlin/eu/inqudium/tabellarium/BoundedWorkerDispatcher.kt @@ -0,0 +1,292 @@ +package eu.inqudium.tabellarium + +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference + +/** + * The one bounded-queue-plus-single-worker skeleton beneath both + * asynchronous hand-offs of the pipeline: [SendDispatcher] (caller → + * `producer.send`) and [FallbackDispatcher] (any thread → the fallback + * appender). Everything the two have in common lives here exactly once + * - the queue, the worker, the accepting state, the in-flight ownership + * protocol, the death handler, the two-phase close, the reentry mark - + * and the two differences are the two abstract methods: what + * *delivering* an item means ([deliver]) and what happens to an item + * that will never be delivered ([reject]). + * + * Rationale: the two dispatchers used to be near-copies whose KDoc + * cross-referenced each other ("mirrors", "the canonical description + * lives there") - and they diverged in exactly one close-budget detail + * that the 2026-09-07 defect analysis had to repair. A prose promise + * that two pieces of code stay identical does not survive change; one + * class does (finding 1 of the 2026-09-07 architecture review). + * + * ## The protocol, once + * + * - **Hand-off** ([offer]) is O(1) and never blocks: a full queue or a + * dispatcher that no longer accepts rejects the item on the caller. + * The check-then-act window between the `running` test and the offer + * is closed by re-checking and reclaiming the item. + * - **In-flight ownership**: the worker records the item it took off + * the queue in [inFlight]; whoever wins the compare-and-set accounts + * for it exactly once - the worker on delivery success or failure, a + * forced [close] when the worker did not finish in time. Without + * this, precisely the item in flight at a forced shutdown would + * vanish from every accounting. + * - **Worker death** (an [Error] escaping [deliver] - [Exception]s are + * handled in place): leave the accepting state FIRST (with the worker + * gone, anything accepted would strand in a queue nothing drains), + * then reject the in-flight item and everything queued, then surface + * the death through the hook. Later offers are rejected on the + * caller. [workerDied] lets a subclass name the terminal cause. + * - **Two-phase close**: the worker keeps DELIVERING for the whole + * drain budget; only then is it interrupted, with a bounded grace for + * a delivery parked in interruptible I/O. Invariant: the interrupt + * ends the drain (an interrupted worker delivers at most one more + * item), so it must never come before the budget has been used. An + * interrupt of the *closing* thread ends its waits early, never the + * accounting, and is restored before returning. + * - **Reentry mark**: the worker sets the appender's [reentryGuard] + * once for its lifetime, so anything delivered work logs through + * SLF4J from this thread is dropped by [KafkaAppender.append] instead + * of looping back into a queue. + * + * Safety: the worker thread starts in the constructor and therefore + * sees `this` before a subclass has initialized its own state - but it + * touches subclass state only through [deliver]/[reject], which need an + * item, and items can only arrive through [offer] after construction. + * + * @param threadName Name of the daemon worker thread. + * @param queueCapacity Maximum queued items; the bound that keeps + * memory finite and makes overflow a visible + * rejection instead of growth. + * @param drainTimeoutMs Time [close] lets the worker drain by + * delivering before interrupting it. + * @param reentryGuard The appender's per-thread reentry guard; null + * disables the marking (tests). + * @param onWorkerDeath Invoked after a worker death has been accounted + * for, so the owner can report it - a dead worker + * must not masquerade as a merely slow consumer. + */ +internal abstract class BoundedWorkerDispatcher( + threadName: String, + queueCapacity: Int, + private val drainTimeoutMs: Long, + private val reentryGuard: ThreadLocal?, + private val onWorkerDeath: (Throwable) -> Unit, +) : AutoCloseable { + /** Why an item will never be delivered; the subclass decides how to account for it. */ + protected enum class Rejection { + /** The queue was full at hand-off time. */ + QUEUE_FULL, + + /** Offered after [close] or after a worker death; see [workerDied] for which. */ + NOT_ACCEPTING, + + /** In flight or queued when the worker died. */ + WORKER_DEATH, + + /** Still in flight or queued when the close budget expired. */ + SHUTDOWN_REMAINDER, + + /** [deliver] threw, and no forced close had claimed the item first. */ + DELIVERY_FAILED, + } + + private val queue: LinkedBlockingQueue = LinkedBlockingQueue(queueCapacity) + + /** The item the worker has taken off the queue but not yet finished delivering; see the class KDoc. */ + private val inFlight = AtomicReference() + + @Volatile + private var running = true + + /** + * True once the worker died: the dispatcher has permanently lost its + * only worker and can never deliver again. Lets a subclass tell a + * post-death rejection from a post-close one. + */ + @Volatile + protected var workerDied: Boolean = false + private set + + /** + * Ensures the close sequence runs exactly once: a second [close] + * (the appender's stop may be invoked repeatedly during context + * teardown) must not re-account the remaining queue or re-join the + * worker. + */ + private val closeExecuted = AtomicBoolean(false) + + private val worker: Thread + + init { + worker = + Thread(::runWorker, threadName).apply { + isDaemon = true + setUncaughtExceptionHandler { _, throwable -> + workerDied = true + running = false + inFlight.getAndSet(null)?.let { reject(it, Rejection.WORKER_DEATH) } + while (true) { + val item = queue.poll() ?: break + reject(item, Rejection.WORKER_DEATH) + } + onWorkerDeath(throwable) + } + start() + } + } + + /** Current queue depth, for the gauges a subclass registers. */ + protected fun queueSize(): Int = queue.size + + /** + * Hands [item] to the worker. Returns `true` if it was queued, + * `false` if it was rejected on the caller (full queue, or no + * longer accepting) - in which case [reject] has already run. + */ + protected fun offer(item: T): Boolean { + if (!running) { + reject(item, Rejection.NOT_ACCEPTING) + return false + } + if (!queue.offer(item)) { + reject(item, Rejection.QUEUE_FULL) + return false + } + // Close the check-then-act window against close() and against + // the death handler: if either finished its final drain between + // the running check and the offer, the item would be neither + // delivered nor accounted for. Re-check and reclaim. + if (!running && queue.remove(item)) { + reject(item, Rejection.NOT_ACCEPTING) + return false + } + return true + } + + override fun close() { + if (!closeExecuted.compareAndSet(false, true)) { + return + } + running = false + // Phase 1: graceful drain. The worker's poll(100, MS) wakes up on + // its next timeout, sees running=false, enters the drain loop and + // keeps delivering until the queue is empty - with the whole + // budget (see the class KDoc for why the interrupt must wait). + var interrupted = false + try { + worker.join(drainTimeoutMs) + } catch (_: InterruptedException) { + interrupted = true + } + if (worker.isAlive) { + // Phase 2: forced exit. Interrupt to wake the worker from + // poll() or from an interruptible delivery, and give the + // interrupt a bounded grace to take effect. Whatever the + // worker is still doing after that (parked in + // non-interruptible I/O) is its own problem now; it is a + // daemon thread, so the JVM can still exit. + worker.interrupt() + // CAUTION: Thread.join(0) means "wait forever", not "do not + // wait" - the grace is a positive constant. Skip the wait if + // we were interrupted ourselves. + if (!interrupted) { + try { + worker.join(INTERRUPT_GRACE_MS) + } catch (_: InterruptedException) { + interrupted = true + } + } + } + // Claim the in-flight item (exactly once via the compare-and-set; + // if the surviving worker still completes the delivery, its own + // CAS fails and nothing is accounted twice - the conservative + // direction), then everything still queued. Drain rather than + // read queue.size, so the items are released and cannot be + // re-accounted by a later call. + inFlight.getAndSet(null)?.let { reject(it, Rejection.SHUTDOWN_REMAINDER) } + while (true) { + val item = queue.poll() ?: break + reject(item, Rejection.SHUTDOWN_REMAINDER) + } + if (interrupted) { + Thread.currentThread().interrupt() + } + } + + private fun runWorker() { + // Set once - the worker never legitimately logs through the appender. + reentryGuard?.set(true) + while (running) { + val item = + try { + queue.poll(100, TimeUnit.MILLISECONDS) + } catch (_: InterruptedException) { + // Forced shutdown: exit immediately rather than pull + // another item that could then park in delivery as a + // ghost; close() accounts for what remains. + Thread.currentThread().interrupt() + return + } ?: continue + deliverGuarded(item) + if (Thread.currentThread().isInterrupted) { + return + } + } + // Graceful drain: running=false, no interrupt. Keep delivering; + // close() waits for this within its budget. + while (true) { + val item = queue.poll() ?: return + deliverGuarded(item) + if (Thread.currentThread().isInterrupted) { + return + } + } + } + + private fun deliverGuarded(item: T) { + inFlight.set(item) + try { + deliver(item) + inFlight.compareAndSet(item, null) + } catch (e: Exception) { + // A failing delivery must not kill the worker: account for + // the item (unless a forced close already claimed it) and + // keep going. An InterruptedException converted from a + // blocking delivery is the shutdown signal - preserve it. + if (inFlight.compareAndSet(item, null)) { + reject(item, Rejection.DELIVERY_FAILED) + } + if (e is InterruptedException) { + Thread.currentThread().interrupt() + } + } + } + + /** Delivers one item on the worker thread. May block; may throw (the item is then rejected as [Rejection.DELIVERY_FAILED]). */ + protected abstract fun deliver(item: T) + + /** + * Accounts for an item that will never be delivered. Called on the + * caller's thread (hand-off rejections), the worker (delivery + * failure, death) or the closing thread (shutdown remainder); must + * not block and must tolerate being called from any of them. + */ + protected abstract fun reject( + item: T, + rejection: Rejection, + ) + + companion object { + /** + * How long [close] waits after interrupting the worker for the + * interrupt to take effect before accounting for the remainder + * itself. Comes on top of the drain budget. + */ + const val INTERRUPT_GRACE_MS: Long = 500 + } +} diff --git a/src/main/kotlin/eu/inqudium/tabellarium/FallbackDispatcher.kt b/src/main/kotlin/eu/inqudium/tabellarium/FallbackDispatcher.kt index 4cc77e8..2ceb717 100644 --- a/src/main/kotlin/eu/inqudium/tabellarium/FallbackDispatcher.kt +++ b/src/main/kotlin/eu/inqudium/tabellarium/FallbackDispatcher.kt @@ -2,16 +2,13 @@ package eu.inqudium.tabellarium import ch.qos.logback.classic.spi.ILoggingEvent import ch.qos.logback.core.Appender -import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong -import java.util.concurrent.atomic.AtomicReference /** * Decouples invocation of the fallback [Appender] from caller threads - * primarily the Kafka producer's I/O thread, which must never block on - * downstream logging. + * downstream logging. A [BoundedWorkerDispatcher] whose delivery is + * `doAppend` and whose every rejection is a counted drop. * * ## Why this exists * @@ -22,13 +19,12 @@ import java.util.concurrent.atomic.AtomicReference * conditions) would block the Kafka I/O thread. Since the Kafka client * has a single I/O thread per producer, blocking it stalls all * subsequent in-flight callbacks, including the `producer.send` calls - * that the application's hot path performs. + * that the send workers perform. * * This dispatcher inserts a single-consumer queue between the Kafka * callback (and the hot-path's synchronous fallback path) and the * actual fallback appender. The caller [enqueue]s in O(1) without - * blocking; a dedicated worker thread drains the queue and calls - * `doAppend`. + * blocking; the worker drains the queue and calls `doAppend`. * * ## Drop policy * @@ -44,28 +40,19 @@ import java.util.concurrent.atomic.AtomicReference * thread (the problem this dispatcher exists to solve). * - An unbounded queue would grow until OOM. * - * ## Lifecycle - * - * - Construction starts the worker thread immediately. The thread is - * marked daemon so it does not prevent JVM shutdown. - * - [close] signals the worker to stop and lets it drain the queue by - * delivering for up to [shutdownTimeoutMs] milliseconds; only then is - * the worker interrupted, with a short bounded grace for a delivery - * parked in `doAppend`. Events still queued or in flight after that - * are dropped (counted in [droppedEventCount]). Invariant: the drain - * gets the whole budget - the interrupt ends the drain, so it must - * never come before the budget has been used. + * The same counting applies to every other way an event can miss the + * appender: `doAppend` throwing (surfacing the exception itself would + * be log-storm-prone, and this class has no status manager - so the + * loss is counted, then swallowed), a worker death, and the remainder + * of a [close] whose budget expired. * * ## Threading and self-logging * - * The worker marks itself with the appender's [reentryGuard] for its - * entire lifetime, exactly like the [SendDispatcher] worker: a fallback - * appender that logs through SLF4J per delivered event would otherwise - * feed each such log back through the root logger into - * [KafkaAppender.append], on to Kafka and - while Kafka is down - back - * into this very queue, a loop that saturates both queues for the - * duration of an outage. With the mark, [KafkaAppender.append] drops - * events raised on this thread. + * The worker carries the appender's reentry guard: a fallback appender + * that logs through SLF4J per delivered event would otherwise feed each + * such log back through the root logger into [KafkaAppender.append], + * on to Kafka and - while Kafka is down - back into this very queue, a + * loop that saturates both queues for the duration of an outage. * * @param fallbackAppender The appender to which events are dispatched. * @param queueCapacity Maximum number of events in flight. Default 1024 @@ -74,46 +61,31 @@ import java.util.concurrent.atomic.AtomicReference * tolerance for brief fallback slowness. * @param shutdownTimeoutMs Time allowed in [close] for the worker to * drain by delivering. Default 5 seconds; the - * bounded interrupt grace comes on top. - * @param reentryGuard The appender's per-thread reentry guard; the - * worker sets it once at startup. Null disables - * the marking (tests). - * @param onWorkerDeath Invoked when the worker thread dies from a - * [Throwable] the delivery loop does not handle - * (an [Error] such as OOM - [Exception]s are - * handled in place). This is the canonical - * description of the death-handler protocol: - * leave the accepting state FIRST (with the - * worker gone, anything accepted would strand in - * a queue nothing ever drains), then account for - * the in-flight event plus everything queued, - * then invoke this hook. The appender reports the - * death to the status manager so a dead worker - * does not masquerade as a merely slow fallback - * appender. + * skeleton's bounded interrupt grace comes on + * top. + * @param reentryGuard The appender's per-thread reentry guard; null + * disables the marking (tests). + * @param onWorkerDeath Invoked after a worker death was accounted for + * (in-flight and queued events counted as + * dropped); the appender reports it to the status + * manager so a dead worker does not masquerade as + * a merely slow fallback appender. */ internal class FallbackDispatcher( private val fallbackAppender: Appender, private val queueCapacity: Int = DEFAULT_QUEUE_CAPACITY, - private val shutdownTimeoutMs: Long = DEFAULT_SHUTDOWN_TIMEOUT_MS, - private val reentryGuard: ThreadLocal? = null, - private val onWorkerDeath: (Throwable) -> Unit = {}, -) : AutoCloseable { - private val queue: LinkedBlockingQueue = LinkedBlockingQueue(queueCapacity) + shutdownTimeoutMs: Long = DEFAULT_SHUTDOWN_TIMEOUT_MS, + reentryGuard: ThreadLocal? = null, + onWorkerDeath: (Throwable) -> Unit = {}, +) : BoundedWorkerDispatcher( + threadName = "kafka-appender-fallback-dispatcher", + queueCapacity = queueCapacity, + drainTimeoutMs = shutdownTimeoutMs, + reentryGuard = reentryGuard, + onWorkerDeath = onWorkerDeath, + ) { private val droppedCount = AtomicLong(0) - /** - * The event the worker has taken off the queue but not yet finished - * delivering. Owned via compare-and-set: exactly one party accounts - * for it. The worker clears it on delivery success (nothing counted) - * or counts it as dropped when `doAppend` throws; a forced [close] - * claims and counts it when the worker did not finish in time. - * Without this, precisely the event in flight at a forced shutdown - * would vanish from the loss accounting - neither delivered nor - * counted as dropped. - */ - private val inFlight = AtomicReference() - /** * Pluggable metrics hook. Defaults to [KafkaAppenderMetrics.NO_OP]; * the appender replaces it via [setMetrics] when a Micrometer @@ -132,49 +104,9 @@ internal class FallbackDispatcher( */ fun setMetrics(metrics: KafkaAppenderMetrics) { this.metrics = metrics - // Re-register the gauges so they bind to the new - // implementation (typically a Micrometer-backed one). The - // queue size supplier is a method reference that always - // reads the current queue state. - metrics.registerFallbackQueueGauges(queueSize = queue::size, capacity = queueCapacity) + metrics.registerFallbackQueueGauges(queueSize = ::queueSize, capacity = queueCapacity) } - @Volatile - private var running = true - - /** - * Ensures the close sequence runs exactly once: a second [close] - * (the appender's stop may be invoked repeatedly during context - * teardown) must not re-count the remaining queue as dropped or - * re-join the worker. - */ - private val closeExecuted = AtomicBoolean(false) - - private val worker: Thread = - Thread(::runWorker, "kafka-appender-fallback-dispatcher").apply { - isDaemon = true - // An Error escaping the delivery loop kills the worker. - // Death-handler protocol (rationale in the onWorkerDeath - // param KDoc): leave the accepting state FIRST, then count - // the in-flight event plus everything queued as dropped, - // then surface the death. Later enqueue calls see - // running=false and count on the caller. - setUncaughtExceptionHandler { _, throwable -> - running = false - val m = metrics - inFlight.getAndSet(null)?.let { - droppedCount.incrementAndGet() - m.fallbackDispatcherDropped() - } - while (queue.poll() != null) { - droppedCount.incrementAndGet() - m.fallbackDispatcherDropped() - } - onWorkerDeath(throwable) - } - start() - } - /** * Number of events lost by this dispatcher: the queue was full when * [enqueue] was called, the fallback appender's `doAppend` threw, @@ -193,171 +125,18 @@ internal class FallbackDispatcher( * - `false` if the queue was full (event dropped) or the dispatcher * has been [close]d. */ - fun enqueue(event: ILoggingEvent): Boolean { - if (!running) { - droppedCount.incrementAndGet() - metrics.fallbackDispatcherDropped() - return false - } - val accepted = queue.offer(event) - if (!accepted) { - droppedCount.incrementAndGet() - metrics.fallbackDispatcherDropped() - return false - } - // Close the check-then-act window against close(): if the - // dispatcher was closed between the running check above and the - // offer, the event may have been added after close() finished its - // final drain accounting - it would then be neither delivered nor - // counted. Re-check and, if we can still pull our own event back - // out, count it as dropped ourselves. - if (!running && queue.remove(event)) { - droppedCount.incrementAndGet() - metrics.fallbackDispatcherDropped() - return false - } - return true - } - - override fun close() { - if (!closeExecuted.compareAndSet(false, true)) { - // Close already ran; nothing left to account for. - return - } - running = false - // Phase 1: graceful drain. The worker's poll(100, MS) wakes up - // on its next timeout, sees running=false, enters the - // drain-on-close loop and keeps DELIVERING until the queue is - // empty. It gets the whole shutdown budget for that - the - // interrupt below ends the drain (an interrupted worker delivers - // at most one more event), so sending it early would throw away - // the rest of the budget together with everything still queued. - // Same two-phase shape as SendDispatcher.close. - // - // An interrupt of the closing thread (e.g. an expiring container - // shutdown budget) must not abort the teardown half-way: the - // joins are wrapped, the forced cleanup and the loss accounting - // below still run without further blocking waits, and the - // interrupt flag is restored before returning. - var interrupted = false - try { - worker.join(shutdownTimeoutMs) - } catch (_: InterruptedException) { - interrupted = true - } + fun enqueue(event: ILoggingEvent): Boolean = offer(event) - if (worker.isAlive) { - // Phase 2: forced exit. The budget is used up; interrupt to - // wake the worker from poll() or from an interruptible - // doAppend, and give the interrupt a short bounded grace to - // take effect. Whatever the worker is still doing after that - // (parked in non-interruptible I/O) is its own problem now. - worker.interrupt() - // CAUTION: Thread.join(0) means "wait forever", not "do not - // wait" - a Java API trap; the grace is a positive constant. - // Skip the wait if we were interrupted ourselves and accept - // that the worker may outlive us; it is a daemon thread, so - // the JVM can still exit. - if (!interrupted) { - try { - worker.join(INTERRUPT_GRACE_MS) - } catch (_: InterruptedException) { - interrupted = true - } - } - } - val m = metrics - // Claim the event the worker is still processing (or abandoned - // mid-delivery): from this point on it counts as dropped exactly - // once. Should the surviving worker still complete the delivery, - // its own compare-and-set fails and nothing is double-counted - - // the conservative direction for a loss metric. - inFlight.getAndSet(null)?.let { - droppedCount.incrementAndGet() - m.fallbackDispatcherDropped() - } - // Any remaining queued events are dropped on shutdown. Drain and - // count in one pass (rather than reading queue.size) so the events - // are actually released and cannot be re-counted by a later call. - while (queue.poll() != null) { - droppedCount.incrementAndGet() - m.fallbackDispatcherDropped() - } - if (interrupted) { - Thread.currentThread().interrupt() - } + override fun deliver(item: ILoggingEvent) { + fallbackAppender.doAppend(item) } - private fun runWorker() { - // Mark this thread for the appender's reentry guard: anything the - // fallback appender logs through SLF4J from inside doAppend now - // happens here, and append() must drop it. Set once - the worker - // never legitimately logs through the appender. - reentryGuard?.set(true) - while (running) { - val event = - try { - queue.poll(100, TimeUnit.MILLISECONDS) - } catch (_: InterruptedException) { - // Shutdown signal received. Exit immediately rather than - // pulling another event from the queue: an in-flight - // event that then blocks the worker in doAppend would - // become a "ghost" - counted neither as delivered nor - // as dropped. The close() method counts everything that - // remains in the queue as dropped, which is the correct - // semantics once we exit here. - Thread.currentThread().interrupt() - return - } ?: continue - - deliver(event) - if (Thread.currentThread().isInterrupted) { - // Forced shutdown arrived while delivering. Stop here; - // close() drains and counts whatever remains queued. - return - } - } - // running=false but no interrupt: graceful shutdown path. - // Drain whatever remains in the queue using non-blocking poll; - // anything still queued or in flight at close() time is counted - // as dropped by close() itself. - while (true) { - val event = queue.poll() ?: return - deliver(event) - if (Thread.currentThread().isInterrupted) { - return - } - } - } - - /** - * Delivers one event to the fallback appender, keeping the - * [inFlight] ownership protocol: on success the slot is cleared - * without counting; when `doAppend` throws, the event is lost and - * counted as dropped - unless a forced [close] already claimed and - * counted it, in which case the compare-and-set fails and the event - * is not double-counted. - */ - private fun deliver(event: ILoggingEvent) { - inFlight.set(event) - try { - fallbackAppender.doAppend(event) - inFlight.compareAndSet(event, null) - } catch (e: Exception) { - // If the fallback throws, the event is gone - surfacing the - // exception itself would be log-storm-prone, and the appender - // has no status manager from this internal class. Account for - // the loss, then swallow. - if (inFlight.compareAndSet(event, null)) { - droppedCount.incrementAndGet() - metrics.fallbackDispatcherDropped() - } - if (e is InterruptedException) { - // Preserve the shutdown signal a blocking appender may - // have converted into an exception. - Thread.currentThread().interrupt() - } - } + override fun reject( + item: ILoggingEvent, + rejection: Rejection, + ) { + droppedCount.incrementAndGet() + metrics.fallbackDispatcherDropped() } companion object { @@ -366,13 +145,5 @@ internal class FallbackDispatcher( /** Default time allowed in close() for the worker to drain by delivering, in milliseconds. */ const val DEFAULT_SHUTDOWN_TIMEOUT_MS: Long = 5000 - - /** - * How long [close] waits after interrupting the worker for the - * interrupt to take effect (a delivery parked in an interruptible - * `doAppend` unblocks) before counting the remainder as dropped - * itself. Mirrors [SendDispatcher]'s grace. - */ - private const val INTERRUPT_GRACE_MS: Long = 500 } } diff --git a/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt b/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt index ca12888..e0d56b4 100644 --- a/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt +++ b/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt @@ -88,10 +88,12 @@ import kotlin.concurrent.withLock * active class) drains into the single fallback queue of the same * default capacity; on a stop during an outage, overflow beyond that * is dropped and counted. - * - **Restart.** `start()` after `stop()` rebuilds the pipeline - * against the still-attached fallback appender (restarting it) and - * re-arms the one-shot hot-path error report. Metrics are not - * rebound automatically - call [bindMeterRegistry] again. + * - **No restart.** `start()` after `stop()` is refused with an error + * (ADR-0004): Logback never restarts an appender instance - a + * reconfiguration stops the old ones and builds new ones - and the + * appender follows that lifecycle instead of carrying every per-life + * resource (fallback, breakers, metrics binding, error guard) across + * a second start. * * ## Why UnsynchronizedAppenderBase * @@ -312,8 +314,8 @@ class KafkaAppender : * Guards the teardown in [stop] so a repeated stop (Logback may call * it more than once during context teardown) does not re-run the * close sequence - re-closing the dispatcher would double-count its - * remaining queue as dropped and re-emit the drop warning. Reset in - * [start] in case the appender is ever restarted. + * remaining queue as dropped and re-emit the drop warning. Never + * reset: once stopped, [start] refuses (ADR-0004). */ private val stopExecuted = AtomicBoolean(false) @@ -328,24 +330,21 @@ class KafkaAppender : addWarn("KafkaAppender is already started; ignoring repeated start().") return } + if (stopExecuted.get()) { + // Rationale: a stopped appender has released its fallback, + // its breakers' history, its metrics binding and its one-shot + // error guard; making all of that come back symmetrically is + // a lifecycle nobody asked for - Logback itself replaces + // instances instead of restarting them (ADR-0004). + addError( + "KafkaAppender cannot be started again after stop() (ADR-0004): Logback replaces " + + "appender instances on reconfiguration - create a new instance instead.", + ) + return + } if (!validateConfiguration()) { return // addError was already called for each failure } - // Restart symmetry: a previous stop() stopped the fallback - // appender (keeping it attached) and latched the one-shot error - // report; both are per-lifecycle and start fresh here. - stopExecuted.set(false) - firstHotPathErrorLogged.set(false) - fallbackAppender?.let { fallback -> - if (!fallback.isStarted) { - try { - fallback.start() - } catch (e: Exception) { - addError("Failed to restart the fallback appender '${fallback.name}' (${e.javaClass.name}): ${e.message}", e) - return - } - } - } // Start the encoder BEFORE the pipeline exists: encoders are // self-contained, so a failing encoder.start() aborts the @@ -470,18 +469,6 @@ class KafkaAppender : }, ) } - // Rationale: the breaker registry lives as long as the appender, - // so on a restart the sender would look up the SAME breakers - // the previous life left behind - possibly OPEN against a - // cluster the operator has since replaced. Everything else in - // the pipeline is rebuilt fresh; the breakers follow suit by - // being reset to CLOSED (identity kept, so the metrics binding's - // per-breaker consumers stay valid). A first start finds none. - registry.activeTopicClasses.forEach { topicClass -> - circuitBreakerRegistry - .find(ResilientMessageSender.circuitBreakerName(topicClass)) - .ifPresent { breaker -> breaker.reset() } - } val sender = ResilientMessageSender( producerRegistry = registry, @@ -800,10 +787,10 @@ class KafkaAppender : // Stop the attached fallback appender. Logback may or may not // hold its own reference to it; calling stop here guarantees its // file handles and worker threads are released even if no other - // path closes it. Deliberately NOT detached: the slot must - // survive for a restart (start() starts it again), and the - // AppenderAttachable contract's detachAndStopAllAppenders stays - // available to callers who really want the slot cleared. + // path closes it. Not detached: the stopped appender stays + // inspectable through the AppenderAttachable accessors, and + // detachAndStopAllAppenders remains available to callers who + // want the slot cleared. try { fallbackAppender?.stop() } catch (e: Exception) { @@ -817,43 +804,25 @@ class KafkaAppender : } /** - * Closes all send dispatchers concurrently and waits for them within - * one shared budget. Each [SendDispatcher.close] is itself bounded + * Closes all send dispatchers concurrently within one shared budget + * ([ParallelClose]). Each [SendDispatcher.close] is itself bounded * (drain timeout plus interrupt grace), so the closer threads always - * finish; the join budget only adds scheduling margin. An interrupt - * of the stopping thread ends the wait early - the daemon closer - * threads complete on their own - and is restored before returning. + * finish; the join budget only adds scheduling margin. */ private fun closeSendDispatchersInParallel() { - if (sendDispatchers.isEmpty()) return - val closers = - sendDispatchers.map { (topicClass, dispatcher) -> - Thread({ - try { - dispatcher.close() - } catch (e: Exception) { - addWarn("Error closing send dispatcher for $topicClass: ${e.message}", e) + ParallelClose.runWithin( + budgetMs = SEND_DISPATCHER_CLOSE_BUDGET_MS, + tasks = + sendDispatchers.map { (topicClass, dispatcher) -> + "tabellarium-send-dispatcher-close-${topicClass.tag}" to { + try { + dispatcher.close() + } catch (e: Exception) { + addWarn("Error closing send dispatcher for $topicClass: ${e.message}", e) + } } - }, "tabellarium-send-dispatcher-close-${topicClass.tag}").apply { - isDaemon = true - start() - } - } - var interrupted = false - val deadlineNanos = System.nanoTime() + SEND_DISPATCHER_CLOSE_BUDGET_MS * 1_000_000 - for (closer in closers) { - val remainingMs = (deadlineNanos - System.nanoTime()) / 1_000_000 - if (remainingMs <= 0) break - try { - closer.join(remainingMs) - } catch (_: InterruptedException) { - interrupted = true - break - } - } - if (interrupted) { - Thread.currentThread().interrupt() - } + }, + ) } // -- Public API: metrics integration -------------------------------- @@ -861,8 +830,9 @@ class KafkaAppender : /** * Whether a [bindMeterRegistry] binding is currently in place. The * [KafkaAppenderMetricsBinding] decides on this - not on appender - * identity - so a restarted instance (whose stop() unbound the - * metrics) is bound again on the next `bindAppenders()` call. + * identity - so an appender whose earlier bind failed is bound on + * the next `bindAppenders()` call and a bound one is never bound + * twice. */ internal val isMeterRegistryBound: Boolean get() = bindLock.withLock { metricsBindings.isBound } diff --git a/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBinding.kt b/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBinding.kt index 9c2b5f9..78975c6 100644 --- a/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBinding.kt +++ b/src/main/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBinding.kt @@ -75,9 +75,8 @@ import java.util.IdentityHashMap * harnesses or context-reload scenarios) result in only one bind per * appender: an appender that already has a metrics binding is * skipped. The decision is made on the appender's own bound state, - * not on instance identity - so a restarted appender (whose `stop()` - * unbound its metrics) is bound again on the next call, and a bind - * that failed is retried. + * not on instance identity - so a bind that failed is retried on the + * next call and a manual bind is never duplicated. * * ## Logback reconfiguration * @@ -90,8 +89,7 @@ import java.util.IdentityHashMap * in place (`LoggerContextListener.onStart` runs only for the initial * start, `onReset` before the new appenders exist), so this class * cannot rebind automatically. Compatibility: after a reconfiguration - * (and equally after a programmatic restart of an appender) the - * metrics stay dark until [bindAppenders] is called again - it is + * the metrics stay dark until [bindAppenders] is called again - it is * public and idempotent for exactly this purpose (e.g. from an * application-side `LoggerContextListener` that defers to the next * scheduler tick, or from an operations endpoint). @@ -148,9 +146,7 @@ open class KafkaAppenderMetricsBinding( } if (appender.isMeterRegistryBound) { // Already bound (by a previous call or manually); the - // appender's own state is the source of truth, so a - // restarted instance - unbound by its stop() - is not - // mistaken for a bound one. + // appender's own state is the source of truth. continue } try { diff --git a/src/main/kotlin/eu/inqudium/tabellarium/ParallelClose.kt b/src/main/kotlin/eu/inqudium/tabellarium/ParallelClose.kt new file mode 100644 index 0000000..0f3404f --- /dev/null +++ b/src/main/kotlin/eu/inqudium/tabellarium/ParallelClose.kt @@ -0,0 +1,57 @@ +package eu.inqudium.tabellarium + +/** + * Runs independent close tasks concurrently and waits for all of them + * within one overall budget - the shape both parallel teardowns of the + * pipeline need ([ProducerRegistry.close] for the producers, + * [KafkaAppender.stop] for the send dispatchers), written once. + * + * Rationale: closing N resources sequentially stacks their individual + * timeouts (`N × timeout`, up to 40 s for four producers), which + * overruns a Kubernetes `terminationGracePeriodSeconds: 30` and gets + * the later closes killed mid-flight. One closer thread per resource + * with a single shared deadline keeps the total at one timeout plus + * margin regardless of N. + * + * Each task runs on its own daemon thread and must handle its own + * exceptions (report, collect) - an exception escaping a task only + * ends that thread. An interrupt of the calling thread ends the wait + * early, leaves the daemon closers to finish on their own, and is + * restored before returning. CAUTION: `Thread.join(0)` means "wait + * forever" - the loop stops before the remaining budget reaches zero. + */ +internal object ParallelClose { + /** + * @param budgetMs Overall wait for all tasks together. + * @param tasks Thread name to task; names should be distinct for + * thread dumps, the order is the join order. + */ + fun runWithin( + budgetMs: Long, + tasks: List Unit>>, + ) { + if (tasks.isEmpty()) return + val closers = + tasks.map { (name, task) -> + Thread(task, name).apply { + isDaemon = true + start() + } + } + var interrupted = false + val deadlineNanos = System.nanoTime() + budgetMs * 1_000_000 + for (closer in closers) { + val remainingMs = (deadlineNanos - System.nanoTime()) / 1_000_000 + if (remainingMs <= 0) break + try { + closer.join(remainingMs) + } catch (_: InterruptedException) { + interrupted = true + break + } + } + if (interrupted) { + Thread.currentThread().interrupt() + } + } +} diff --git a/src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt b/src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt index 9b080ce..69c872a 100644 --- a/src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt +++ b/src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt @@ -93,40 +93,24 @@ internal class ProducerRegistry private constructor( * as suppressed exceptions) so the caller's warn path can surface * them instead of losing them silently. * - * An interrupt while waiting stops the wait early, restores the - * interrupt flag, and leaves the daemon closer threads to finish on - * their own. + * The parallel wait itself is [ParallelClose] - shared with the + * appender's dispatcher teardown - including its interrupt handling. */ override fun close() { val failures = ConcurrentLinkedQueue>() - val closers = - producersByClass.map { (topicClass, producer) -> - Thread({ - try { - producer.close(closeTimeout) - } catch (e: Exception) { - failures += topicClass to e + ParallelClose.runWithin( + budgetMs = closeTimeout.toMillis() + JOIN_MARGIN.toMillis(), + tasks = + producersByClass.map { (topicClass, producer) -> + "tabellarium-producer-close-${topicClass.name.lowercase()}" to { + try { + producer.close(closeTimeout) + } catch (e: Exception) { + failures += topicClass to e + } } - }, "tabellarium-producer-close-${topicClass.name.lowercase()}").apply { - isDaemon = true - start() - } - } - var interrupted = false - val deadlineNanos = System.nanoTime() + closeTimeout.toNanos() + JOIN_MARGIN.toNanos() - for (closer in closers) { - val remainingMs = (deadlineNanos - System.nanoTime()) / 1_000_000 - if (remainingMs <= 0) break - try { - closer.join(remainingMs) - } catch (_: InterruptedException) { - interrupted = true - break - } - } - if (interrupted) { - Thread.currentThread().interrupt() - } + }, + ) if (failures.isNotEmpty()) { val summary = failures.joinToString(separator = "; ") { (topicClass, cause) -> diff --git a/src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt b/src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt index 110d3e5..eb65d59 100644 --- a/src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt +++ b/src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt @@ -1,15 +1,14 @@ package eu.inqudium.tabellarium import ch.qos.logback.classic.spi.ILoggingEvent -import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean -import java.util.concurrent.atomic.AtomicReference /** * Decouples `producer.send` from the logging caller's thread - the * asynchronous heart of the appender's "the sender is never made to - * wait" promise. + * wait" promise. A [BoundedWorkerDispatcher] whose delivery is the + * potentially-blocking send and whose rejections divert to the + * fallback. * * ## Why this exists * @@ -34,41 +33,35 @@ import java.util.concurrent.atomic.AtomicReference * PERFORMANCE delivery. FIFO order per topic class is preserved by * the single worker. * - * ## Overflow and shutdown policy + * ## Diversion policy * - * The queue is **bounded**. When it is full, [dispatch] never blocks: - * the event is diverted to the fallback dispatcher (when configured) - * and counted as [KafkaAppenderMetrics.FallbackReason.QUEUE_FULL]. A - * full queue means Kafka delivery is not keeping up - the fallback is - * the designed escape hatch for exactly that state, and blocking the - * caller would resurrect the problem this class exists to solve. - * - * On [close], the worker first drains the queue gracefully (producers - * are still open - the appender closes send dispatchers before the - * producer registry). If the drain does not finish within the budget, - * the worker is interrupted (a send parked in `max.block.ms` unblocks - * with an `InterruptException`, which the sender's error path routes - * to the fallback) and everything still queued or in flight is - * diverted to the fallback with - * [KafkaAppenderMetrics.FallbackReason.SHUTDOWN] - accounted exactly - * once via the same compare-and-set ownership protocol the - * [FallbackDispatcher] uses for its in-flight event. + * Every rejection of the skeleton becomes a fallback diversion with a + * metric reason: a full queue is `queue.full` (Kafka delivery is not + * keeping up - the fallback is the designed escape hatch for exactly + * that state, and blocking the caller would resurrect the problem this + * class exists to solve); the remainder of a [close] is `shutdown` + * (the drain still sends - the appender closes send dispatchers before + * the producer registry; a send parked in `max.block.ms` unblocks on + * the interrupt with an `InterruptException`, which the sender's error + * path routes itself); a worker death, a failed delivery, and a + * dispatch after a death are `send.error`. Every diversion is + * accounted exactly once via [PendingSend.claim], shared with the + * sender's own diversion paths. * * ## Threading and self-logging * - * The worker thread marks itself with the appender's [reentryGuard] - * ThreadLocal for its entire lifetime: the Kafka client logs - * synchronously on the `producer.send` caller - which is now this + * The worker carries the appender's reentry guard: the Kafka client + * logs synchronously on the `producer.send` caller - which is now this * worker - and those events must be dropped by [KafkaAppender.append] * instead of being fed back into the queue (a feedback loop that * amplifies exactly during broker trouble). * * The [ILoggingEvent] crosses to the worker thread only as the payload * for the *fallback* path - the same cross-thread exposure the - * [FallbackDispatcher] already has today, since the Kafka callback - * thread hands events to it as well. Encoding and enrichment already - * happened on the original caller thread, so MDC and markers were read - * in their native context. + * [FallbackDispatcher] already has, since the Kafka callback thread + * hands events to it as well. Encoding and enrichment already happened + * on the original caller thread, so MDC and markers were read in their + * native context. * * @param topicClass The topic class this dispatcher serves; used for * metrics tagging and the worker thread name. @@ -76,26 +69,19 @@ import java.util.concurrent.atomic.AtomicReference * `messageSender.send(topicClass, ...)`. Injected as * a function so the dispatcher can be tested with * latches instead of a full Kafka pipeline. - * @param fallbackDispatcher Receives diverted events (queue overflow, - * shutdown remainder). Null means "drop" - - * the operator's explicit choice, consistent - * with the rest of the pipeline. - * @param reentryGuard The appender's per-thread reentry guard; the - * worker sets it once at startup. Null disables - * the marking (tests). + * @param fallbackDispatcher Receives diverted events. Null means + * "drop" - the operator's explicit choice, + * consistent with the rest of the pipeline. + * @param reentryGuard The appender's per-thread reentry guard; null + * disables the marking (tests). * @param queueCapacity Maximum queued events. The default matches the * fallback dispatcher's: large enough to absorb * bursts, small enough to bound memory. * @param drainTimeoutMs Time allowed in [close] for the worker to * drain the queue by actually sending. - * @param onWorkerDeath Invoked when the worker thread dies - same - * trigger and death-handler protocol as the - * [FallbackDispatcher] hook (the canonical - * description lives there), except that the - * affected work is diverted to the fallback with - * reason `send.error` instead of drop-counted, - * and later [dispatch] calls divert on the - * caller. The appender reports the death to the + * @param onWorkerDeath Invoked after a worker death was accounted for + * (in-flight and queued work diverted with reason + * `send.error`); the appender reports it to the * status manager so it does not masquerade as a * slow broker. */ @@ -103,11 +89,17 @@ internal class SendDispatcher( private val topicClass: TopicClass, private val sendAction: (PendingSend) -> Unit, private val fallbackDispatcher: FallbackDispatcher?, - private val reentryGuard: ThreadLocal? = null, + reentryGuard: ThreadLocal? = null, private val queueCapacity: Int = DEFAULT_QUEUE_CAPACITY, - private val drainTimeoutMs: Long = DEFAULT_DRAIN_TIMEOUT_MS, - private val onWorkerDeath: (Throwable) -> Unit = {}, -) : AutoCloseable { + drainTimeoutMs: Long = DEFAULT_DRAIN_TIMEOUT_MS, + onWorkerDeath: (Throwable) -> Unit = {}, +) : BoundedWorkerDispatcher( + threadName = "kafka-appender-send-dispatcher-${topicClass.tag}", + queueCapacity = queueCapacity, + drainTimeoutMs = drainTimeoutMs, + reentryGuard = reentryGuard, + onWorkerDeath = onWorkerDeath, + ) { /** * The unit of work handed from the caller to the worker: everything * the send needs, pre-computed on the caller's thread. @@ -153,65 +145,16 @@ internal class SendDispatcher( fun tryClaim(): Boolean = diverted.compareAndSet(false, true) } - private val queue: LinkedBlockingQueue = LinkedBlockingQueue(queueCapacity) - - /** - * The item the worker has taken off the queue but not yet finished - * sending. Same compare-and-set ownership protocol as - * [FallbackDispatcher]: exactly one party accounts for it on a - * forced shutdown. - */ - private val inFlight = AtomicReference() - @Volatile private var metrics: KafkaAppenderMetrics = KafkaAppenderMetrics.NO_OP - @Volatile - private var running = true - - /** - * Set by the worker's uncaught-exception handler: the dispatcher - * has permanently lost its only worker and can never deliver again. - * Distinguishes the terminal diversion reason in [dispatch] - - * `send.error` after a worker death versus `shutdown` after - * [close] - so operators see the real cause instead of a phantom - * shutdown. - */ - @Volatile - private var workerDied = false - - private val closeExecuted = AtomicBoolean(false) - - private val worker: Thread = - Thread(::runWorker, "kafka-appender-send-dispatcher-${topicClass.tag}").apply { - isDaemon = true - // Death-handler protocol as in FallbackDispatcher (the - // canonical rationale lives there): leave the accepting - // state FIRST, then divert the in-flight item and the - // queue (reason send.error), then surface the death via - // onWorkerDeath. - setUncaughtExceptionHandler { _, throwable -> - workerDied = true - running = false - inFlight.getAndSet(null)?.let { - divert(it, KafkaAppenderMetrics.FallbackReason.SEND_ERROR) - } - while (true) { - val item = queue.poll() ?: break - divert(item, KafkaAppenderMetrics.FallbackReason.SEND_ERROR) - } - onWorkerDeath(throwable) - } - start() - } - /** * Replaces the metrics implementation and registers the queue * gauges with it. Called by [KafkaAppender.bindMeterRegistry]. */ fun setMetrics(metrics: KafkaAppenderMetrics) { this.metrics = metrics - metrics.registerSendQueueGauges(topicClass, queueSize = queue::size, capacity = queueCapacity) + metrics.registerSendQueueGauges(topicClass, queueSize = ::queueSize, capacity = queueCapacity) } /** @@ -225,141 +168,46 @@ internal class SendDispatcher( enrichment: EnrichedRecord, originalEvent: ILoggingEvent, ) { - val item = PendingSend(topicName, payload, enrichment, originalEvent) - if (!running) { - divert(item, terminalDiversionReason()) - return - } - if (!queue.offer(item)) { - divert(item, KafkaAppenderMetrics.FallbackReason.QUEUE_FULL) - return - } - // Close the check-then-act window against close() and against - // the worker-death handler, same as FallbackDispatcher.enqueue: - // if either finished its final drain between the running check - // and the offer, the item would be neither sent nor diverted. - // Re-check and reclaim. - if (!running && queue.remove(item)) { - divert(item, terminalDiversionReason()) - } + offer(PendingSend(topicName, payload, enrichment, originalEvent)) } - /** - * Why the dispatcher stopped accepting: a worker death diverts as - * `send.error` (delivery capability was lost to an error), a - * regular [close] as `shutdown`. - */ - private fun terminalDiversionReason(): KafkaAppenderMetrics.FallbackReason = - if (workerDied) { - KafkaAppenderMetrics.FallbackReason.SEND_ERROR - } else { - KafkaAppenderMetrics.FallbackReason.SHUTDOWN - } + override fun deliver(item: PendingSend) { + // ResilientMessageSender.send handles its own error paths; an + // exception here is unexpected and becomes a send.error divert. + sendAction(item) + } - override fun close() { - if (!closeExecuted.compareAndSet(false, true)) { - return - } - running = false - // Two-phase shutdown: the graceful drain (the worker keeps - // SENDING - the producers are still open at this point) gets - // the full budget; only then is the worker interrupted, with a - // short bounded wait for the interrupt to take effect. The - // interrupt handling mirrors [FallbackDispatcher.close]: an - // interrupted closer still runs the forced cleanup and restores - // its flag (finding M-6 in - // docs/assessment/CODE_ANALYSIS-2026-08-28T22-20-43.md). - var interrupted = false - try { - worker.join(drainTimeoutMs) - } catch (_: InterruptedException) { - interrupted = true - } - if (worker.isAlive) { - // A send parked in max.block.ms unblocks with an - // InterruptException; the sender's error path routes that - // event to the fallback itself. - worker.interrupt() - if (!interrupted) { - try { - worker.join(INTERRUPT_GRACE_MS) - } catch (_: InterruptedException) { - interrupted = true + override fun reject( + item: PendingSend, + rejection: Rejection, + ) { + val reason = + when (rejection) { + Rejection.QUEUE_FULL -> { + KafkaAppenderMetrics.FallbackReason.QUEUE_FULL } - } - } - // Claim the in-flight item (exactly-once via CAS; if the worker - // still completes the send, its own CAS fails and nothing is - // diverted twice), then divert everything still queued. - inFlight.getAndSet(null)?.let { - divert(it, KafkaAppenderMetrics.FallbackReason.SHUTDOWN) - } - while (true) { - val item = queue.poll() ?: break - divert(item, KafkaAppenderMetrics.FallbackReason.SHUTDOWN) - } - if (interrupted) { - Thread.currentThread().interrupt() - } - } - private fun runWorker() { - // Mark this thread for the appender's reentry guard: everything - // the Kafka client logs synchronously from inside producer.send - // now happens here, and append() must drop it. Set once - the - // worker never legitimately logs through the appender. - reentryGuard?.set(true) - while (running) { - val item = - try { - queue.poll(100, TimeUnit.MILLISECONDS) - } catch (_: InterruptedException) { - // Forced shutdown: exit immediately; close() diverts - // what remains. - Thread.currentThread().interrupt() - return - } ?: continue - deliver(item) - if (Thread.currentThread().isInterrupted) { - return - } - } - // Graceful drain: running=false, no interrupt. Keep sending - - // the producers are still open, close() waits for this. - while (true) { - val item = queue.poll() ?: return - deliver(item) - if (Thread.currentThread().isInterrupted) { - return - } - } - } + Rejection.SHUTDOWN_REMAINDER -> { + KafkaAppenderMetrics.FallbackReason.SHUTDOWN + } - private fun deliver(item: PendingSend) { - inFlight.set(item) - try { - sendAction(item) - inFlight.compareAndSet(item, null) - } catch (e: Exception) { - // Unexpected: ResilientMessageSender.send handles its own - // error paths internally. Whatever slipped through must not - // kill the worker - divert the event (unless close() already - // claimed it) and keep going. - if (inFlight.compareAndSet(item, null)) { - divert(item, KafkaAppenderMetrics.FallbackReason.SEND_ERROR) - } - if (e is InterruptedException) { - Thread.currentThread().interrupt() - } - } - } + Rejection.WORKER_DEATH, Rejection.DELIVERY_FAILED -> { + KafkaAppenderMetrics.FallbackReason.SEND_ERROR + } - private fun divert( - item: PendingSend, - reason: KafkaAppenderMetrics.FallbackReason, - ) { + // A dispatch after the worker died lost its delivery + // capability to an error; after a regular close it is a + // shutdown - operators see the real cause either way. + Rejection.NOT_ACCEPTING -> { + if (workerDied) { + KafkaAppenderMetrics.FallbackReason.SEND_ERROR + } else { + KafkaAppenderMetrics.FallbackReason.SHUTDOWN + } + } + } // Exactly-once across ALL diversion paths, the sender's - // included - see PendingSend.tryClaimDiversion. + // included - see PendingSend.claim. if (!item.tryClaimDiversion()) { return } @@ -373,12 +221,5 @@ internal class SendDispatcher( /** Default time allowed in [close] for the worker to drain by sending, in milliseconds. */ const val DEFAULT_DRAIN_TIMEOUT_MS: Long = 1000 - - /** - * How long [close] waits after interrupting the worker for the - * interrupt to take effect (a parked send unblocks with an - * InterruptException) before diverting the remainder itself. - */ - private const val INTERRUPT_GRACE_MS: Long = 500 } } diff --git a/src/test/kotlin/eu/inqudium/tabellarium/DocumentationContractTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/DocumentationContractTest.kt new file mode 100644 index 0000000..ecb5794 --- /dev/null +++ b/src/test/kotlin/eu/inqudium/tabellarium/DocumentationContractTest.kt @@ -0,0 +1,91 @@ +package eu.inqudium.tabellarium + +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test +import java.nio.file.Files +import java.nio.file.Path + +/** + * Keeps the configuration guide's numbers in step with the code. The + * guide's defaults quick reference is the canonical statement of the + * "code" defaults (README and KDoc link there instead of restating + * them); this test is what makes it canonical - a changed constant + * fails the build until the table follows, the same anti-drift + * principle the generated coverage and test-evidence pages apply + * (finding 2 of the 2026-09-07 architecture review). + * + * Runs from the module root (Surefire's working directory), where the + * guide lives under `docs/`. + */ +class DocumentationContractTest { + private val guide: List = + Files.readAllLines(Path.of("docs", "config", "kafka-appender-config-guide.md")) + + /** The "Default" cell of the defaults-table row whose "Setting" cell starts with [setting]. */ + private fun defaultOf(setting: String): String { + val row = + guide.singleOrNull { it.startsWith("| $setting") } + ?: error("defaults table row '$setting' not found exactly once in the configuration guide") + return row.split("|")[2].trim() + } + + /** The "Default" cell of the circuit-breaker table row for [property]. */ + private fun breakerDefaultOf(property: String): String { + val row = + guide.singleOrNull { it.startsWith("| `$property`") } + ?: error("circuit-breaker table row '$property' not found exactly once in the configuration guide") + return row.split("|")[2].trim() + } + + @Test + fun `should state the dispatcher and producer budgets exactly as the constants define them`() { + // What is to be tested? Whether the guide's defaults quick + // reference carries the queue capacities and the shutdown + // budgets that the code actually uses. + // How will the test case be deemed successful and why? Successful + // if each "code" row's Default cell contains the value formatted + // from the corresponding constant - so a constant change without + // a table update fails the build. + // Why is it important to test this test case? These numbers drifted + // three times on 2026-09-07 alone (the 200 ms drain window that + // fix 3 removed survived in the guide); operators size + // termination grace periods from this table. + assertThat(defaultOf("Send dispatcher queue capacity (per class)")) + .contains("`${SendDispatcher.DEFAULT_QUEUE_CAPACITY}`") + assertThat(defaultOf("Send dispatcher drain on stop (parallel)")) + .contains("`${SendDispatcher.DEFAULT_DRAIN_TIMEOUT_MS / 1000} s` drain") + assertThat(defaultOf("Fallback dispatcher queue capacity")) + .contains("`${FallbackDispatcher.DEFAULT_QUEUE_CAPACITY}`") + assertThat(defaultOf("Fallback dispatcher shutdown timeout")) + .contains("`${FallbackDispatcher.DEFAULT_SHUTDOWN_TIMEOUT_MS / 1000} s` drain") + .contains("`${BoundedWorkerDispatcher.INTERRUPT_GRACE_MS / 1000.0} s` interrupt grace") + assertThat(defaultOf("Producer close timeout")) + .contains("`${ProducerRegistry.DEFAULT_CLOSE_TIMEOUT.toSeconds()} s`") + } + + @Test + fun `should state the circuit-breaker and throttle defaults exactly as the configuration defines them`() { + // Given: the production breaker configuration + val config = ResilientMessageSender.defaultCircuitBreakerConfig() + + // Then: both guide tables (the breaker section and the quick reference) match it + assertThat(breakerDefaultOf("failureRateThreshold")).isEqualTo("`${config.failureRateThreshold.toInt()}%`") + assertThat(breakerDefaultOf("slidingWindowSize")).isEqualTo("`${config.slidingWindowSize}` calls") + assertThat(breakerDefaultOf("minimumNumberOfCalls")).isEqualTo("`${config.minimumNumberOfCalls}`") + assertThat(breakerDefaultOf("waitDurationInOpenState")) + .isEqualTo("`${config.waitIntervalFunctionInOpenState.apply(1) / 1000}s`") + assertThat(breakerDefaultOf("permittedNumberOfCallsInHalfOpenState")) + .isEqualTo("`${config.permittedNumberOfCallsInHalfOpenState}`") + + assertThat(defaultOf("Circuit breaker: failure-rate threshold")).isEqualTo("`${config.failureRateThreshold.toInt()}%`") + assertThat(defaultOf("Circuit breaker: sliding window / min calls")) + .isEqualTo("`${config.slidingWindowSize}` / `${config.minimumNumberOfCalls}`") + assertThat(defaultOf("Circuit breaker: open-state wait")) + .isEqualTo("`${config.waitIntervalFunctionInOpenState.apply(1) / 1000}s`") + assertThat(defaultOf("Circuit breaker: half-open permitted calls")) + .isEqualTo("`${config.permittedNumberOfCallsInHalfOpenState}`") + assertThat(defaultOf("Half-open probe gap")) + .isEqualTo("`${ResilientMessageSender.DEFAULT_HALF_OPEN_PROBE_GAP.toMillis()} ms`") + assertThat(defaultOf("Partitioning key MDC source")).isEqualTo("`${MessageEnricher.TRACE_ID_MDC_KEY}`") + } +} diff --git a/src/test/kotlin/eu/inqudium/tabellarium/FallbackDispatcherTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/FallbackDispatcherTest.kt index f04080b..87a7dd2 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/FallbackDispatcherTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/FallbackDispatcherTest.kt @@ -22,22 +22,6 @@ class FallbackDispatcherTest { */ private val testContext = LoggerContext() - /** Appender that records the events it receives. */ - private inner class RecordingAppender : AppenderBase() { - val events = mutableListOf() - - init { - context = testContext - start() - } - - override fun append(event: ILoggingEvent) { - synchronized(events) { events += event } - } - - fun eventCount(): Int = synchronized(events) { events.size } - } - /** * Appender that blocks on each append until released. With * [interruptible] = false the block survives the worker interrupt @@ -132,7 +116,7 @@ class FallbackDispatcherTest { @Test fun `should deliver enqueued events to the fallback appender on the worker thread`() { // Given - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val dispatcher = FallbackDispatcher(recorder) try { // When @@ -200,7 +184,7 @@ class FallbackDispatcherTest { @Test fun `should drain remaining events when closed gracefully`() { // Given - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val dispatcher = FallbackDispatcher(recorder) // When: enqueue events, then close immediately @@ -215,7 +199,7 @@ class FallbackDispatcherTest { @Test fun `should mark events enqueued after close as dropped`() { // Given - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val dispatcher = FallbackDispatcher(recorder) dispatcher.close() diff --git a/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBindingTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBindingTest.kt index 05e850f..960cb77 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBindingTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBindingTest.kt @@ -34,27 +34,6 @@ import org.springframework.context.event.ContextRefreshedEvent class KafkaAppenderMetricsBindingTest { // -- Test fixtures -------------------------------------------------- - /** - * Minimal encoder: formatted message → UTF-8 bytes. Enough to - * exercise the appender hot path without pulling in Logstash. - */ - private class TestEncoder : EncoderBase() { - override fun encode(event: ILoggingEvent): ByteArray = event.formattedMessage.toByteArray(Charsets.UTF_8) - - override fun headerBytes(): ByteArray = ByteArray(0) - - override fun footerBytes(): ByteArray = ByteArray(0) - } - - /** - * Producer factory returning auto-completing MockProducers, so - * `producer.send` callbacks fire synchronously. No real Kafka - * cluster involved. - */ - private class MockProducerFactory : ProducerFactory { - override fun create(properties: Map): Producer = MockProducer(true, FixedZeroPartitioner(), ByteArraySerializer(), ByteArraySerializer()) - } - private lateinit var loggerContext: LoggerContext private lateinit var appender: KafkaAppender @@ -79,7 +58,7 @@ class KafkaAppenderMetricsBindingTest { context = loggerContext name = "TEST_KAFKA" encoder = - TestEncoder().also { + MessageBytesEncoder().also { it.context = loggerContext it.start() } @@ -90,7 +69,7 @@ class KafkaAppenderMetricsBindingTest { topicMapping = TopicMappingConfig().apply { defaultTopic = "default.topic" } // Inject the mock producer factory so start() succeeds without // a real Kafka cluster. Same hook KafkaAppenderTest uses. - producerFactory = MockProducerFactory() + producerFactory = RecordingProducerFactory() start() } @@ -211,55 +190,6 @@ class KafkaAppenderMetricsBindingTest { } } - @Test - fun `should bind a restarted appender again when bindAppenders is called`() { - // What is to be tested? Whether the binding decides on the - // appender's own bound state rather than on instance - // identity: an appender that was stopped (which unbinds its - // metrics) and started again is the same instance, and a - // manual bindAppenders() call - the documented rebind path - - // must bind it again instead of skipping it as "known". - // How will the test case be deemed successful and why? Successful - // if after a stop/start cycle the appender's counters are - // gone from the registry, and after bindAppenders() a - // hot-path event moves the accepted counter by exactly one - // again. - // Why is it important to test this test case? Finding R2-4 of - // the 2026-09-07 follow-up: with the identity set, the - // restarted appender stayed dark although the documentation - // named bindAppenders() as the way to relight it. - - ApplicationContextRunner() - .withUserConfiguration(MeterRegistryConfig::class.java, BindingConfig::class.java) - .run { ctx -> - val registry = ctx.getBean(MeterRegistry::class.java) - val binding = ctx.getBean(KafkaAppenderMetricsBinding::class.java) - assertThat(registry.find("kafka.appender.events.accepted").counters()).isNotEmpty - - // When: the appender is restarted (stop unbinds) and - // the binding is asked again - appender.stop() - assertThat(registry.find("kafka.appender.events.accepted").counters()).isEmpty() - appender.start() - assertThat(appender.isStarted).isTrue() - binding.bindAppenders() - - // Then: bound again - the counters exist and count - val before = - registry - .find("kafka.appender.events.accepted") - .counters() - .sumOf { it.count() } - appender.doAppend(loggingEvent()) - val after = - registry - .find("kafka.appender.events.accepted") - .counters() - .sumOf { it.count() } - assertThat(after - before).isEqualTo(1.0) - } - } - @Test fun `should bind only once even if the context publishes refresh multiple times`() { // What is to be tested? Whether the binding is idempotent diff --git a/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderTest.kt index 3f88ee8..7cf5593 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderTest.kt @@ -45,67 +45,6 @@ import java.util.concurrent.atomic.AtomicReference class KafkaAppenderTest { // -- Test fixtures -------------------------------------------------- - private class TestEncoder : EncoderBase() { - val encodedEvents = mutableListOf() - - override fun encode(event: ILoggingEvent): ByteArray { - encodedEvents += event - return event.formattedMessage.toByteArray(Charsets.UTF_8) - } - - override fun headerBytes(): ByteArray = ByteArray(0) - - override fun footerBytes(): ByteArray = ByteArray(0) - } - - private class ThrowingEncoder : EncoderBase() { - override fun encode(event: ILoggingEvent): ByteArray = throw RuntimeException("simulated encoder failure") - - override fun headerBytes(): ByteArray = ByteArray(0) - - override fun footerBytes(): ByteArray = ByteArray(0) - } - - /** - * Stateless encoder for tests that append from many threads - * concurrently: unlike [TestEncoder] it records nothing, so the - * test harness itself introduces no unsynchronized shared state - * (TestEncoder's recording list is a plain ArrayList). - */ - private class StatelessEncoder : EncoderBase() { - override fun encode(event: ILoggingEvent): ByteArray = event.formattedMessage.toByteArray(Charsets.UTF_8) - - override fun headerBytes(): ByteArray = ByteArray(0) - - override fun footerBytes(): ByteArray = ByteArray(0) - } - - private class TestProducerFactory : ProducerFactory { - val createdProducers = mutableListOf>() - val createdWithProperties = mutableListOf>() - - override fun create(properties: Map): Producer { - val mock = MockProducer(true, FixedZeroPartitioner(), ByteArraySerializer(), ByteArraySerializer()) - createdProducers += mock - createdWithProperties += properties - return mock - } - } - - private class RecordingAppender : AppenderBase() { - // Synchronized: the asynchronous-dispatch test appends from the - // dispatcher worker thread while the test thread polls. - val events: MutableList = Collections.synchronizedList(mutableListOf()) - - init { - start() - } - - override fun append(event: ILoggingEvent) { - events += event - } - } - /** * Every appender a test built; stopped after the test (stop() is * idempotent) so no dispatcher worker, fallback worker or @@ -120,13 +59,13 @@ class KafkaAppenderTest { } private fun newAppender( - encoder: Encoder? = TestEncoder(), + encoder: Encoder? = RecordingEncoder(), component: String = "test-service", cmdbId: String = "CMDB-TEST", environment: String = "test", defaultTopic: String = "default.topic", debug: Boolean = false, - producerFactory: ProducerFactory = TestProducerFactory(), + producerFactory: ProducerFactory = RecordingProducerFactory(), fallback: Appender? = null, kafkaProducerProperties: String = "${ProducerConfig.BOOTSTRAP_SERVERS_CONFIG}=test:9092", ): KafkaAppender = @@ -256,7 +195,7 @@ class KafkaAppenderTest { @Test fun `should start the encoder when starting the appender`() { // Given - val encoder = TestEncoder() + val encoder = RecordingEncoder() val appender = newAppender(encoder = encoder) // When @@ -284,7 +223,7 @@ class KafkaAppenderTest { // showed up in capacity planning. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) // When @@ -310,7 +249,7 @@ class KafkaAppenderTest { // which topic class) a connection belongs to. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory, component = "checkout-service") // When @@ -336,7 +275,7 @@ class KafkaAppenderTest { // deployment whose component name contains a space. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory, component = "My Service (prod)") // When @@ -350,7 +289,7 @@ class KafkaAppenderTest { @Test fun `should let an operator-supplied client id win`() { // Given: the operator pins client.id in kafkaProducerProperties - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender( producerFactory = factory, @@ -392,7 +331,7 @@ class KafkaAppenderTest { // configuration and the actual broker behavior. // Given: an AUDIT mapping and a conflicting operator acks=1 - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender( producerFactory = factory, @@ -439,7 +378,7 @@ class KafkaAppenderTest { // dormant TECHNICAL producer running. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender( producerFactory = factory, @@ -480,7 +419,7 @@ class KafkaAppenderTest { // activation itself is part of the operator contract. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) appender.topicMapping.addMapping( TopicMappingEntry().apply { @@ -809,8 +748,8 @@ class KafkaAppenderTest { // the system is least able to absorb it. // Given - val encoder = TestEncoder() - val factory = TestProducerFactory() + val encoder = RecordingEncoder() + val factory = RecordingProducerFactory() val fallback = RecordingAppender() val appender = newAppender( @@ -840,7 +779,7 @@ class KafkaAppenderTest { @Test fun `should deliver events from ordinary threads`() { // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) appender.start() @@ -870,7 +809,7 @@ class KafkaAppenderTest { // guard only ever suppresses the producer's own logging. // Given: operator pins the generic client.id "app" - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender( producerFactory = factory, @@ -910,7 +849,7 @@ class KafkaAppenderTest { // the appender into a black hole. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender( producerFactory = factory, @@ -936,8 +875,8 @@ class KafkaAppenderTest { @Test fun `should encode and send the event when appended`() { // Given - val encoder = TestEncoder() - val factory = TestProducerFactory() + val encoder = RecordingEncoder() + val factory = RecordingProducerFactory() val appender = newAppender(encoder = encoder, producerFactory = factory) appender.start() @@ -972,7 +911,7 @@ class KafkaAppenderTest { // nothing to Kafka - every event diverted as encoder.error. // Given: an appender with a fallback recorder - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val fallback = RecordingAppender() val appender = newAppender(producerFactory = factory, fallback = fallback) appender.start() @@ -1035,7 +974,7 @@ class KafkaAppenderTest { // them. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) // Note: start() not called @@ -1226,7 +1165,7 @@ class KafkaAppenderTest { // an external stop() could ever reach. // Given: an encoder whose start() throws - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val throwingStartEncoder = object : EncoderBase() { override fun start(): Unit = throw IllegalStateException("simulated encoder start failure") @@ -1266,7 +1205,7 @@ class KafkaAppenderTest { // does. // Given: breaker wiring that fails after the registry exists - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val failingBreakerRegistry = object : CircuitBreakerRegistry by ResilientMessageSender.defaultCircuitBreakerRegistry() { override fun circuitBreaker(name: String): CircuitBreaker = throw IllegalStateException("simulated breaker wiring failure") @@ -1289,7 +1228,7 @@ class KafkaAppenderTest { @Test fun `should close all producers when stopping`() { // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) appender.start() @@ -1303,7 +1242,7 @@ class KafkaAppenderTest { @Test fun `should stop the encoder when stopping`() { // Given - val encoder = TestEncoder() + val encoder = RecordingEncoder() val appender = newAppender(encoder = encoder) appender.start() @@ -1364,7 +1303,7 @@ class KafkaAppenderTest { // When appender.stop() - // Then: stopped, but still attached (a restart starts it again) + // Then: stopped, and still inspectable through the slot assertThat(fallback.isStarted).isFalse() assertThat(appender.fallbackAppender).isSameAs(fallback) } @@ -1423,7 +1362,7 @@ class KafkaAppenderTest { val fallback = RecordingAppender() val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = selfLoggingFactory, fallback = fallback, ) @@ -1458,7 +1397,7 @@ class KafkaAppenderTest { // permanently silenced application thread. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) appender.start() @@ -1474,144 +1413,45 @@ class KafkaAppenderTest { } @Nested - inner class `Restart` { - @Test - fun `should keep the fallback path and re-arm the error report after a stop and start cycle`() { - // What is to be tested? Restart symmetry: start() after stop() - // must rebuild the pipeline against the still-attached - // fallback appender (restarting it) and re-arm the one-shot - // hot-path error report, so a restarted appender is not a - // silently fallback-less one. - // How will the test case be deemed successful and why? Successful - // if, after stop() and start(), a diverted event still - // reaches the fallback appender, the fallback reports - // started, and the hot-path error is reported once more. - // Why is it important to test this test case? Before the fix, - // stop() detached the fallback slot; the restarted pipeline - // dropped every diversion although the operator's XML still - // named the fallback - and the latched error guard hid the - // first error of the new lifecycle. - - // Given: a started-then-stopped appender with a fallback - val fallback = RecordingAppender() - val appender = newAppender(encoder = ThrowingEncoder(), fallback = fallback) - appender.start() - appender.doAppend(newTestLoggingEvent(message = "first life")) - appender.stop() - assertThat(fallback.events.map { it.formattedMessage }).containsExactly("first life") - assertThat(fallback.isStarted).isFalse() - - // When: restarted, and an event diverts again - appender.start() - assertThat(appender.isStarted).isTrue() - assertThat(fallback.isStarted).isTrue() - appender.doAppend(newTestLoggingEvent(message = "second life")) - appender.stop() - - // Then: the fallback path survived the cycle, and the error - // was reported once per lifecycle - assertThat(fallback.events.map { it.formattedMessage }).containsExactly("first life", "second life") - assertThat(appender.statusMessages().filter { it.contains("Hot path error") }).hasSize(2) - } - + inner class `No restart` { @Test - fun `should reset a breaker left open by the previous life when restarted`() { - // What is to be tested? Whether a restart rebuilds the - // resilience state as well as the pipeline: the breaker - // registry lives as long as the appender, so without a reset - // the new sender would inherit an OPEN breaker from the - // previous life and divert against a possibly replaced, - // healthy cluster for the rest of the open-state wait. + fun `should refuse to start again after stop`() { + // What is to be tested? The lifecycle decision of ADR-0004: a + // stopped appender is not restarted - start() after stop() + // is refused with a named error, no new producers are + // created, and the appender stays stopped. // How will the test case be deemed successful and why? Successful - // if a breaker forced OPEN before stop() reads CLOSED after - // start() and the first event of the new life reaches the - // producer instead of the fallback. - // Why is it important to test this test case? The restart - // support (finding 4 of the 2026-09-07 analysis) made this - // path real; finding R2-5 of the follow-up found the - // carried-over state. - - // Given: a started appender whose TECHNICAL breaker is OPEN - val factory = TestProducerFactory() + // if after stop() a second start() leaves isStarted false, + // the producer count unchanged, and a status error naming + // the ADR and the alternative (a new instance) in the + // status manager; an event appended afterwards goes nowhere. + // Why is it important to test this test case? Logback replaces + // appender instances on reconfiguration instead of restarting + // them; a same-instance restart only ever comes from + // application code, and silently rebuilding half a pipeline + // for it (the pre-ADR behavior) lost the fallback path. + + // Given: a started-then-stopped appender + val factory = RecordingProducerFactory() val fallback = RecordingAppender() val appender = newAppender(producerFactory = factory, fallback = fallback) appender.start() - val breaker = - appender.circuitBreakerRegistry - .circuitBreaker(ResilientMessageSender.circuitBreakerName(TopicClass.TECHNICAL)) - breaker.transitionToOpenState() appender.stop() + val producersAfterFirstLife = factory.createdProducers.size - // When: restarted + // When: started again appender.start() - appender.doAppend(newTestLoggingEvent(message = "second life")) - appender.stop() + appender.doAppend(newTestLoggingEvent(message = "after refusal")) - // Then: the breaker was reset and the event was sent, not diverted - assertThat(breaker.state).isEqualTo(CircuitBreaker.State.CLOSED) - assertThat( - factory.createdProducers - .last() - .history() - .map { String(it.value()) }, - ).containsExactly("second life") + // Then: refused, nothing rebuilt, nothing delivered anywhere + assertThat(appender.isStarted).isFalse() + assertThat(factory.createdProducers).hasSize(producersAfterFirstLife) + assertThat(appender.statusMessages()) + .anyMatch { it.contains("cannot be started again") && it.contains("ADR-0004") } assertThat(fallback.events).isEmpty() } } - @Nested - inner class `Fallback worker reentry guard` { - @Test - fun `should drop events a fallback appender logs from its own delivery instead of looping them`() { - // What is to be tested? Whether an event that the fallback - // appender itself raises from inside doAppend (a - // third-party appender logging through SLF4J per delivered - // event) is dropped by the reentry guard on the fallback - // worker instead of re-entering the pipeline. - // How will the test case be deemed successful and why? Successful - // if, with the encoder failing (so every event diverts), a - // fallback that re-logs each delivered event through the - // appender ends up with exactly the application's own - // events - and the pipeline terminates. Without the guard - // every fallback delivery would spawn a new event, and the - // fallback would keep receiving events until the test - // stopped the appender. - // Why is it important to test this test case? During a Kafka - // outage this loop saturates both queues and crowds out the - // genuine events - exactly when the fallback is the only - // remaining record. - - // Given: a fallback appender that logs back through the appender - var appenderRef: KafkaAppender? = null - val delivered = Collections.synchronizedList(mutableListOf()) - val reLoggingFallback = - object : AppenderBase() { - init { - context = LoggerContext() - start() - } - - override fun append(event: ILoggingEvent) { - delivered += event.formattedMessage - checkNotNull(appenderRef).doAppend( - newTestLoggingEvent(message = "fallback said: ${event.formattedMessage}"), - ) - } - } - val appender = newAppender(encoder = ThrowingEncoder(), fallback = reLoggingFallback) - appenderRef = appender - appender.start() - - // When: the application logs three events, all diverting - repeat(3) { appender.doAppend(newTestLoggingEvent(message = "app-$it")) } - - // Then: the fallback received exactly the application's events; - // its own re-logged events were dropped on the worker - appender.stop() - assertThat(delivered).containsExactly("app-0", "app-1", "app-2") - } - } - @Nested inner class `Repeated start` { @Test @@ -1631,7 +1471,7 @@ class KafkaAppenderTest { // until process exit. // Given - val factory = TestProducerFactory() + val factory = RecordingProducerFactory() val appender = newAppender(producerFactory = factory) appender.start() val producersAfterFirstStart = factory.createdProducers.size @@ -1723,7 +1563,7 @@ class KafkaAppenderTest { val factory = BlockingProducerFactory() val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = factory, ) appender.start() @@ -1763,7 +1603,7 @@ class KafkaAppenderTest { val factory = BlockingProducerFactory(blockedClasses = setOf("audit")) val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = factory, ) appender.topicMapping.addMapping( @@ -1814,7 +1654,7 @@ class KafkaAppenderTest { val fallback = RecordingAppender() val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = factory, fallback = fallback, ) @@ -1893,7 +1733,7 @@ class KafkaAppenderTest { val fallback = RecordingAppender() val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = throwingBlockedFactory, fallback = fallback, ) @@ -2010,7 +1850,7 @@ class KafkaAppenderTest { val fallback = RecordingAppender() val appender = newAppender( - encoder = StatelessEncoder(), + encoder = MessageBytesEncoder(), producerFactory = factory, fallback = fallback, ) diff --git a/src/test/kotlin/eu/inqudium/tabellarium/KafkaBrokerIntegrationTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/KafkaBrokerIntegrationTest.kt index 4a63400..71f49c4 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/KafkaBrokerIntegrationTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/KafkaBrokerIntegrationTest.kt @@ -38,15 +38,6 @@ import java.util.concurrent.TimeUnit */ @Tag("integration") class KafkaBrokerIntegrationTest { - /** Minimal encoder so the payload assertion is byte-exact. */ - private class PlainTextEncoder : EncoderBase() { - override fun encode(event: ILoggingEvent): ByteArray = event.formattedMessage.toByteArray(Charsets.UTF_8) - - override fun headerBytes(): ByteArray = ByteArray(0) - - override fun footerBytes(): ByteArray = ByteArray(0) - } - @Test fun `should deliver a TECHNICAL and an AUDIT record through a real broker`() { // What is to be tested? The central external system boundary @@ -87,7 +78,7 @@ class KafkaBrokerIntegrationTest { val appender = KafkaAppender().apply { context = LoggerContext() - encoder = PlainTextEncoder() + encoder = MessageBytesEncoder() component = "integration-test" cmdbId = "CMDB-IT" environment = "it" diff --git a/src/test/kotlin/eu/inqudium/tabellarium/ProducerRegistryTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/ProducerRegistryTest.kt index 1cf9552..ae37705 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/ProducerRegistryTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/ProducerRegistryTest.kt @@ -18,23 +18,6 @@ class ProducerRegistryTest { private fun newBuilder(base: Map = baseProperties) = ProducerPropertiesBuilder(base) - /** - * Test factory that records every invocation. Returns auto-completing - * [MockProducer]s so send() calls (if any) succeed without configuring - * a Cluster. - */ - private class RecordingFactory : ProducerFactory { - val createdProducers = mutableListOf>() - val receivedProperties = mutableListOf>() - - override fun create(properties: Map): Producer { - receivedProperties += properties - val mock = MockProducer(true, FixedZeroPartitioner(), ByteArraySerializer(), ByteArraySerializer()) - createdProducers += mock - return mock - } - } - /** * Test producer that throws on [close], used to verify the registry's * per-producer try/catch in [ProducerRegistry.close]. @@ -58,7 +41,7 @@ class ProducerRegistryTest { ProducerRegistry.create( propertiesBuilder = newBuilder(), activeTopicClasses = emptySet(), - producerFactory = RecordingFactory(), + producerFactory = RecordingProducerFactory(), ) }.isInstanceOf(IllegalArgumentException::class.java) .hasMessageContaining("At least one active topic class") @@ -67,7 +50,7 @@ class ProducerRegistryTest { @Test fun `should create exactly one producer per active topic class`() { // Given - val factory = RecordingFactory() + val factory = RecordingProducerFactory() // When val registry = @@ -98,7 +81,7 @@ class ProducerRegistryTest { // compliance. // Given: base sets a value that AUDIT will override - val factory = RecordingFactory() + val factory = RecordingProducerFactory() val builder = newBuilder( mapOf( @@ -115,8 +98,8 @@ class ProducerRegistryTest { ) // Then: the factory received the enforced value - assertThat(factory.receivedProperties).hasSize(1) - assertThat(factory.receivedProperties[0]) + assertThat(factory.createdWithProperties).hasSize(1) + assertThat(factory.createdWithProperties[0]) .containsEntry(ProducerConfig.ACKS_CONFIG, "all") } @@ -140,7 +123,7 @@ class ProducerRegistryTest { TopicClass.FUNCTIONAL, TopicClass.TECHNICAL, ), - producerFactory = RecordingFactory(), + producerFactory = RecordingProducerFactory(), ) // Then @@ -156,7 +139,7 @@ class ProducerRegistryTest { @Test fun `should return the producer instance that was created for the given topic class`() { // Given - val factory = RecordingFactory() + val factory = RecordingProducerFactory() val registry = ProducerRegistry.create( propertiesBuilder = newBuilder(), @@ -178,7 +161,7 @@ class ProducerRegistryTest { ProducerRegistry.create( propertiesBuilder = newBuilder(), activeTopicClasses = setOf(TopicClass.AUDIT), - producerFactory = RecordingFactory(), + producerFactory = RecordingProducerFactory(), ) // When / Then @@ -244,7 +227,7 @@ class ProducerRegistryTest { @Test fun `should close all producers when the registry is closed`() { // Given - val factory = RecordingFactory() + val factory = RecordingProducerFactory() val registry = ProducerRegistry.create( propertiesBuilder = newBuilder(), diff --git a/src/test/kotlin/eu/inqudium/tabellarium/ResilientMessageSenderTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/ResilientMessageSenderTest.kt index 838f835..d907462 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/ResilientMessageSenderTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/ResilientMessageSenderTest.kt @@ -33,30 +33,6 @@ class ResilientMessageSenderTest { ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "broker:9092", ) - /** - * Test factory that returns MockProducers with the given autoComplete mode. - * autoComplete=true → the send callback fires immediately with success; - * autoComplete=false → the test must call mockProducer.completeNext() or - * mockProducer.errorNext(...) to trigger the callback. - */ - private class TestFactory( - private val autoComplete: Boolean = true, - /** - * Optional wrapper around each created MockProducer - the seam - * for producer doubles that model client behavior MockProducer - * lacks (e.g. the synchronous error callback). - */ - private val wrap: (MockProducer) -> Producer = { it }, - ) : ProducerFactory { - val createdProducers = mutableListOf>() - - override fun create(properties: Map): Producer { - val mock = MockProducer(autoComplete, FixedZeroPartitioner(), ByteArraySerializer(), ByteArraySerializer()) - createdProducers += mock - return wrap(mock) - } - } - /** * Models the Kafka client's ApiException path (kafka-clients 4.x, * `KafkaProducer.doSend`): metadata not available within @@ -78,24 +54,6 @@ class ResilientMessageSenderTest { } } - /** - * Test appender that records every event it receives. Started in its - * init block because AppenderBase.doAppend() is a no-op for unstarted - * appenders. The list is synchronized: the fallback dispatcher's - * worker thread appends while the test thread polls. - */ - private class RecordingAppender : AppenderBase() { - val events: MutableList = Collections.synchronizedList(mutableListOf()) - - init { - start() - } - - override fun append(event: ILoggingEvent) { - events += event - } - } - /** * Capturing [KafkaAppenderMetrics] for tests: records every hook * call in a thread-safe list so assertions can inspect what the @@ -173,7 +131,7 @@ class ResilientMessageSenderTest { cbRegistry: CircuitBreakerRegistry = CircuitBreakerRegistry.ofDefaults(), wrapProducer: (MockProducer) -> Producer = { it }, ): SenderContext { - val factory = TestFactory(autoComplete, wrapProducer) + val factory = RecordingProducerFactory(autoComplete, wrapProducer) val registry = ProducerRegistry.create( propertiesBuilder = ProducerPropertiesBuilder(baseProperties), @@ -198,7 +156,7 @@ class ResilientMessageSenderTest { private data class SenderContext( val sender: ResilientMessageSender, - val factory: TestFactory, + val factory: RecordingProducerFactory, val circuitBreakerRegistry: CircuitBreakerRegistry, val fallback: RecordingAppender?, val registry: ProducerRegistry, diff --git a/src/test/kotlin/eu/inqudium/tabellarium/SendDispatcherTest.kt b/src/test/kotlin/eu/inqudium/tabellarium/SendDispatcherTest.kt index deb987e..69f515d 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/SendDispatcherTest.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/SendDispatcherTest.kt @@ -19,20 +19,6 @@ class SendDispatcherTest { private val testContext = LoggerContext() - /** Fallback recorder; the list is synchronized because the fallback worker writes while the test polls. */ - private inner class RecordingAppender : AppenderBase() { - val events: MutableList = Collections.synchronizedList(mutableListOf()) - - init { - context = testContext - start() - } - - override fun append(event: ILoggingEvent) { - events += event - } - } - /** Metrics recorder for the fallback reasons the dispatcher emits. */ private class RecordingMetrics : KafkaAppenderMetrics by KafkaAppenderMetrics.NO_OP { val fallbackReasons: MutableList = @@ -218,7 +204,7 @@ class SendDispatcherTest { // Given: worker pinned, capacity 1 val release = CountDownLatch(1) val entered = CountDownLatch(1) - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val metrics = RecordingMetrics() val dispatcher = SendDispatcher( @@ -275,7 +261,7 @@ class SendDispatcherTest { // Given val sent = AtomicInteger(0) - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val dispatcher = SendDispatcher( topicClass = TopicClass.TECHNICAL, @@ -315,7 +301,7 @@ class SendDispatcherTest { val release = CountDownLatch(1) val entered = CountDownLatch(1) val sendReturned = AtomicBoolean(false) - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val metrics = RecordingMetrics() val fallbackDispatcher = newFallback(recorder) val dispatcher = @@ -371,7 +357,7 @@ class SendDispatcherTest { @Test fun `should divert events dispatched after close to the fallback`() { // Given - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val metrics = RecordingMetrics() val dispatcher = SendDispatcher( @@ -411,7 +397,7 @@ class SendDispatcherTest { // Given: a send action that dies with an Error val death = AtomicReference() - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val dispatcher = SendDispatcher( topicClass = TopicClass.TECHNICAL, @@ -456,7 +442,7 @@ class SendDispatcherTest { val entered = CountDownLatch(1) val release = CountDownLatch(1) val death = AtomicReference() - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val metrics = RecordingMetrics() val dispatcher = SendDispatcher( @@ -522,7 +508,7 @@ class SendDispatcherTest { val release = CountDownLatch(1) val entered = CountDownLatch(1) val lateClaim = AtomicReference() - val recorder = RecordingAppender() + val recorder = RecordingAppender(testContext) val fallbackDispatcher = newFallback(recorder) val dispatcher = SendDispatcher( diff --git a/src/test/kotlin/eu/inqudium/tabellarium/TestSupport.kt b/src/test/kotlin/eu/inqudium/tabellarium/TestSupport.kt new file mode 100644 index 0000000..0557d00 --- /dev/null +++ b/src/test/kotlin/eu/inqudium/tabellarium/TestSupport.kt @@ -0,0 +1,98 @@ +package eu.inqudium.tabellarium + +import ch.qos.logback.classic.LoggerContext +import ch.qos.logback.classic.spi.ILoggingEvent +import ch.qos.logback.core.encoder.EncoderBase +import org.apache.kafka.clients.producer.MockProducer +import org.apache.kafka.clients.producer.Producer +import org.apache.kafka.common.serialization.ByteArraySerializer + +/* + * Shared test support for the pipeline tests. One recorder, one + * producer factory, three encoders - the fixtures every test class + * used to declare privately (four recorders, ~17 producer doubles, + * six encoders across the suite; finding 4 of the 2026-09-07 + * architecture review). Test-specific producer doubles that model one + * behavior (blocking, self-logging, synchronous callback errors) stay + * next to the test that needs them; the shared factory takes them as + * the `wrap` function. + */ + +/** + * A started, thread-safe recording appender: the fallback recorder of + * the suite. Events are read from the test thread while a dispatcher + * worker appends - the copy-on-write list of [ThreadSafeListAppender] + * gives every read a fully published snapshot. + * + * @param context Logback context; set so `doAppend` emits no + * "No context given" status noise. A fresh one per + * recorder is fine for the programmatic tests. + */ +internal class RecordingAppender( + context: LoggerContext = LoggerContext(), +) : ThreadSafeListAppender() { + init { + this.context = context + start() + } + + fun eventCount(): Int = events.size +} + +/** + * [ProducerFactory] returning [MockProducer]s and recording what it + * created and with which properties. + * + * @param autoComplete `true` completes every send synchronously with + * success; `false` defers to `completeNext()` / + * `errorNext(...)` on the test's side. + * @param wrap Optional decorator around each created mock - the seam + * for producer doubles that model client behavior + * `MockProducer` lacks. + */ +internal class RecordingProducerFactory( + private val autoComplete: Boolean = true, + private val wrap: (MockProducer) -> Producer = { it }, +) : ProducerFactory { + val createdProducers = mutableListOf>() + val createdWithProperties = mutableListOf>() + + override fun create(properties: Map): Producer { + val mock = MockProducer(autoComplete, FixedZeroPartitioner(), ByteArraySerializer(), ByteArraySerializer()) + createdProducers += mock + createdWithProperties += properties + return wrap(mock) + } +} + +/** Stateless encoder: the formatted message as UTF-8 bytes. Safe for concurrent appends. */ +internal open class MessageBytesEncoder : EncoderBase() { + override fun encode(event: ILoggingEvent): ByteArray = event.formattedMessage.toByteArray(Charsets.UTF_8) + + override fun headerBytes(): ByteArray = ByteArray(0) + + override fun footerBytes(): ByteArray = ByteArray(0) +} + +/** + * [MessageBytesEncoder] that additionally records every encoded event. + * The list is a plain `ArrayList`: encoding runs on the caller's + * thread, so use it only from single-threaded tests. + */ +internal class RecordingEncoder : MessageBytesEncoder() { + val encodedEvents = mutableListOf() + + override fun encode(event: ILoggingEvent): ByteArray { + encodedEvents += event + return super.encode(event) + } +} + +/** Encoder whose `encode` always throws - the hot-path failure injector. */ +internal class ThrowingEncoder : EncoderBase() { + override fun encode(event: ILoggingEvent): ByteArray = throw RuntimeException("simulated encoder failure") + + override fun headerBytes(): ByteArray = ByteArray(0) + + override fun footerBytes(): ByteArray = ByteArray(0) +} diff --git a/src/test/kotlin/eu/inqudium/tabellarium/ThreadSafeListAppender.kt b/src/test/kotlin/eu/inqudium/tabellarium/ThreadSafeListAppender.kt index ac97022..f2b8289 100644 --- a/src/test/kotlin/eu/inqudium/tabellarium/ThreadSafeListAppender.kt +++ b/src/test/kotlin/eu/inqudium/tabellarium/ThreadSafeListAppender.kt @@ -14,9 +14,10 @@ import java.util.concurrent.CopyOnWriteArrayList * snapshot. * * Instantiated reflectively by Joran in the XML round-trip tests, so - * the class needs its public no-arg constructor. + * the class needs its public no-arg constructor; the programmatic + * tests use its started subclass [RecordingAppender]. */ -internal class ThreadSafeListAppender : AppenderBase() { +internal open class ThreadSafeListAppender : AppenderBase() { val events = CopyOnWriteArrayList() override fun append(event: ILoggingEvent) { From 97c4ff71810290041290baaaf8db86e396db89bc Mon Sep 17 00:00:00 2001 From: dirkjink Date: Mon, 7 Sep 2026 20:48:31 +0200 Subject: [PATCH 2/2] Add the 2026-09-07T20-23 architecture review with the remediation status Fourth-round appropriateness analysis at ffede8b (0 Critical, 0 High, 3 Medium, 3 Low, 2 systemic patterns); every finding carries its status line referencing the fix commit, finding 3 recorded as ADR-0004. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_015GfdGp7eUrjJKBvcJUx3q2 --- ...ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md | 138 ++++++++++++++++++ 1 file changed, 138 insertions(+) create mode 100644 docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md diff --git a/docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md b/docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md new file mode 100644 index 0000000..f188ed8 --- /dev/null +++ b/docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md @@ -0,0 +1,138 @@ +# Architecture & Appropriateness Analysis: tabellarium - 4th round + +1. Identification of the Codebase + - **Repository:** `https://github.com/Inqudium/tabellarium.git` + - **Commit-Hash:** `ffede8bcd89477d76e73d42035eab9c391a6570f` (full; merge of PR #16, `refs/heads/main`, working tree clean) + - **Reference (Branch/Tag):** `refs/heads/main` / nearest tag: `v1.0.0` (`revision` = `1.0.1-SNAPSHOT`) + - **Ticket / work item:** [MISSING - please supply] (no ticket named in the request; add one for a complete audit record) +2. Scope of the Analysis + - **Included:** `./src/main/kotlin/eu/inqudium/tabellarium/` (17 production files, 2 122 code lines + 2 680 comment lines), the build descriptor `./pom.xml` and `./.mvn/jvm.config`, the architectural self-description (`./README.md`, `./docs/index.md`, `./docs/api-module.md`, `./docs/config/kafka-appender-config-guide.md`, `./docs/adr/`, `./CONTRIBUTING.md`), and the separate JMH module `./benchmarks/` (its POM and README, for boundary and lifecycle coherence). + - **Test code:** **included as an analysis subject** - the test-architecture section of Phase 2 is applied in full depth (fixture design, seams, pyramid shape), *not* correctness or flakiness (those are the defect analyses of the same day: `./docs/assessment/CODE_ANALYSIS-2026-09-07T19-09-00.md` and its `.R2.md`). Counting basis for the test topology: `./src/test/kotlin/` (21 files) and `./src/test/java/` (3 fuzz targets). + - **Excluded:** `./target/`, `./.github/workflows/` (read only for their count and purpose), the Python generators under `./.github/scripts/` (reviewed in round 3 for proportionality, unchanged since), earlier assessment documents except as adoption basis (see methodology). +3. Analysis Environment & Tools + - **Target Environment:** artifact targets Java 21; build requires JDK 24+ (local Oracle JDK 26.0.1, CI JDK 25); Kotlin 2.4.10 + - **Build system:** Apache Maven 3.9.15, Spring Boot parent 4.1.1 (BOM/plugin parent only - the module is a Logback appender library; Spring is an optional dependency for one metrics-binding class) + - **Analysis tools used:** manual reading of all in-scope files (the production code is fully known from this session's two defect-analysis passes and re-read here through the appropriateness lens), grep-based counting for the pattern bases (declaration inventory, fixture duplication, cross-references, deadline loops), a code/comment line classifier (small Python script over `./src/main/kotlin/`), `mvn dependency:tree` (round-2 defect pass) for the dependency picture; no static-analysis product beyond the project's own gates (ktlint, CodeQL in CI). No code executed for this analysis. +4. Placement & Output + - **Working directory (workdir):** `/home/dirk/IdeaProjects/tabellarium` (absolute reference point; all relative paths refer to it) + - **Report output path:** `./docs/assessment/ARCHITECTURE_REVIEW-2026-09-07T20-23-00.md` (relative to the workdir; prefix + ISO 8601 timestamp) + - **Predecessor architecture reviews (read-only history, verified before adoption):** `./docs/assessment/ARCHITECTURE_REVIEW-2026-08-28T18-53-39.md` (round 1), `./docs/assessment/ARCHITECTURE_REVIEW-2026-08-29T00-34-03.md` (round 2, commit chain up to `6d2a025`), `./docs/assessment/ARCHITECTURE_REVIEW-2026-08-29T02-16-27.md` (round 3, commit `6841d0fe7bdd59e57451ed6183059601274c3fcd`) + - **Scope root (relative to the workdir):** `./src/main/kotlin/` (production), `./src/test/` (tests), `./` (build and documentation surfaces) + - **Path convention for findings:** `:` - line numbers refer to commit `ffede8b…` + +**Status update: 2026-09-07, fix commit `698e2b3` (report base: `ffede8b`).** All six findings are addressed - five fixed, finding 3 decided in the "refuse restart" direction and recorded as ADR-0004. The full suite (267 tests: the three restart tests are gone, two documentation-contract tests are new; `mvn verify` offline incl. ktlint) is green; the benchmark module compiles against the refactored library (locally and, from now on, in CI); the documentation-contract test was mutation-checked (a changed guide number fails it). Details per finding in the **Status** entries in section 5. + +--- + +## 1. Executive Summary + +Tabellarium's architecture fits its problem, and this fourth round confirms the trajectory the previous three established: the means are proportionate to a resilient, non-blocking Logback-to-Kafka transport with per-class producer policy - bounded queues, one worker per active class, breaker plus throttle, an asynchronous fallback path, a four-type operator surface (ADR-0002), optional dependencies gated consistently, and every operational number published from a build rather than by hand. Nothing here is Critical or High; there is no reactive theater, no home-grown framework, no speculative interface (both interfaces in the module have two real implementations or a real seam), and the one deliberately heavy piece - the own Resilience4j binder - carries a documented, re-verified rationale. The tendency, if any, is a **mild under-engineering of the shared skeleton beneath the two dispatchers** and a **mild over-supply of the lifecycle**: the two dispatchers are near-copies whose divergence the same-day defect analysis had to repair (finding 1), the same rationale and the same numbers live in up to five hand-maintained prose surfaces that drifted three times in one day despite the project's own generated-artifact principle (finding 2), and this session's remediation added a *restart* capability - fallback restart, breaker reset, binding rebind - for which no consumer requirement is on record (finding 3). Three Low findings concern the test-support duplication, the benchmark module standing outside every build, and one documentation gap. The problem baseline still lacks the one input every appropriateness judgment leans on: real consumer load, deployments and the actual use of the four topic classes remain `[MISSING - please supply]`, as in all three previous rounds. + +**Test verdict (tests in scope):** + +1. **Testability of the architecture:** very good and structurally earned - function-shaped seams (`ProducerFactory`, `sendAction`, injected nano clocks, `KafkaAppenderMetrics`, `reentryGuard`), `internal` visibility for test composition, and no synchronous bypass in production code; every score-4/5 unit is verifiable in isolation with hand-written fakes, and the appender itself against `MockProducer` on the real worker path. The only structural residue is the wall-clock-bound shutdown budgets (defect pattern P3), which force shortened real waits instead of an injected clock. +2. **Utilization & coverage (test pyramid):** healthy and used to the full - 269 test methods, 264 in the default offline run (no broker, no Docker, four lightweight `ApplicationContextRunner` uses, three Joran tests with a real unreachable-broker producer), one real-broker Testcontainers proof in its own CI job, four external-contract characterizations behind a profile, three Jazzer targets. Weight sits on fast isolated tests; the orchestrator class carries the largest share (64 of 269), which is the appender's role as composition root, not an inverted pyramid. +3. **Most significant gaps & anomalies:** + - Test support is copy-pasted rather than shared: four private `RecordingAppender`s, ~17 producer-double definitions across four files, six minimal encoders across three files (finding 4) - the round-2 `ListAppender` thread-safety fix already had to be redone in one of the private copies. + - The benchmark module - the project's only performance regression instrument - is compiled by nobody: not in the reactor, not in CI (finding 5), against the repository's own "cannot rot silently" principle. + - No test seam exists for time in the shutdown paths (defect pattern P3, deferred with rationale); the drain-budget tests are wall-clock tests by necessity. + +## 2. Problem Baseline & Methodology + +- **Core domain (re-verified at this commit):** a Logback appender library - encode on the caller, route by SLF4J marker, classify into four producer-policy classes with mandatory overrides for the graded ones, hand off in O(1) to one bounded queue and worker per active class, send through a per-class breaker with half-open throttling, divert what cannot be shipped into an optional fallback appender through a second bounded queue, account every loss, publish metrics on demand. Code mass: 2 122 code lines in 17 files, accompanied by 2 680 comment lines (ratio 1.26; `KafkaAppender` 461/500, `MetricsBindings` 280/124, the smallest units up to 4:1). The metrics subsystem (4 files, ~635 code lines) is ~30 % of the code for an optional feature - accepted in round 2 with a corrected, still-valid rationale. +- **Real requirements & scale:** the documented forces are unchanged and real - callers (incl. Reactor event loops and virtual threads) must never block, memory must stay bounded, classes must fail independently, loss must be visible, shutdown must fit a 30 s termination grace, dependencies must stay optional. The performance analysis of 2026-08-29 *assumed* 10 k events/s and the benchmark report retired every optimization at that profile; the one measured optimization (shared pre-built headers, 160 B/op) was folded in. **Still missing, for the fourth round running:** consumer deployments, production rates, whether FUNCTIONAL/PERFORMANCE are used anywhere, whether any consumer needs a programmatic restart or breaker tuning. Every "is this proportionate?" judgment below is therefore calibrated against the documented forces, not against observed use. +- **Team & process context:** single maintainer; the repository doubles as the reference template for the Inqudium ecosystem's build/CI/quality machinery (documented outside the repo; in-repo the POM and CONTRIBUTING explain each gate) - the 14 Maven plugins, 6 workflows (548 lines), SBOM/OSV, CodeQL, Scorecard, SLSA provenance, Dokka site, Jazzer nightly and three Python generators are therefore a *deliberate* investment with a stated purpose, not accidental heft, and are not reported. Note (no finding, rationale explicit in `./pom.xml:614` and `./CONTRIBUTING.md:9`): the JDK 24+ build floor for a Java-21-targeting library exists solely to keep `mvn verify` warning-free; the cost is a contributor toolchain three majors above the target. +- **Documented architectural intent:** ADR-0001 (comment prefix vocabulary), ADR-0002 (public API = operator surface; extension by widening the XML surface, never by exposing internals), ADR-0003 (Fuzz workflow is the fuzzing signal); the README "Delivery guarantees" section (best-effort transport with visible loss - the round-2 product decision); CONTRIBUTING (lock-free hot path, no synchronous test modes, three-question test rationale, generated evidence); standing decisions in earlier review status entries (own Resilience4j binder for its teardown lifecycle; external-contract test kept until a target repository exists; producer-registry consolidation deferred until a deployment hits the ceiling). **Code versus ADRs: no deviation found** - the surface is exactly the four ADR-0002 types (`KafkaAppender`, `TopicMappingConfig`/`TopicMappingEntry`, `TopicClass`, `KafkaAppenderMetricsBinding`; counting basis: the declaration inventory above, 21 top-level declarations, 4 non-`internal`), the stale breaker-override promise that contradicted ADR-0002 was removed in `5d849c8` today, and the comment vocabulary is in use. +- **Technology coherence:** sound and unchanged - blocking Kafka client behind dedicated platform-thread workers, `UnsynchronizedAppenderBase`, `LinkedBlockingQueue` hand-offs, Resilience4j 2.4, Micrometer/Spring/Logstash all optional and probe-gated; no coroutines, no Reactor, no persistence. The only Spring code is one `@EventListener` helper. +- **Test topology & testability (baseline signal):** 269 test methods; per-test infrastructure load in the default run is a `MockProducer` and hand-built fakes; the heaviest fixtures are latch-pinned producer doubles. Seams are function-shaped and `internal`; no mock library (CONTRIBUTING). Coverage ~91 % (published). Shape: a healthy pyramid with a broad unit base, one broker proof, a fuzz layer. +- **Adoption basis from earlier rounds:** all round-2 and round-3 findings were re-verified as fixed or as standing decisions at this commit (grep: no `synchronous` test mode, no all-open plugin, `internal` seams, broker-stage CI job present, independent profile exclusions, README diagram current). Nothing was adopted unchecked; none is re-reported. +- **Analyzed vs. not analyzed:** all 17 production units read; test architecture read at the fixture level for every test class; build and docs surfaces read; `./benchmarks/` read for boundary coherence, not benchmarked. **Blind spots:** no consumer repositories or runtime data; the two Grafana dashboards under `./docs/metrics/` were not audited; the Python generators unchanged since round 3 and not re-read. + +## 3. Statistics + +| Severity | Count | +|---|---:| +| 🔴 Critical | 0 | +| 🟠 High | 0 | +| 🟡 Medium | 3 | +| 🟢 Low | 3 | +| **Total findings** | **6** | +| **Systemic patterns** | **2** (+ 2 positive counter-patterns recorded for balance) | + +By category: Under-Engineering/Consistency 1, Consistency 2, Over-Engineering (speculative feature) 1, Testability & Test Architecture 1, Dependency & Build Appropriateness 1. + +## 4. Ranking Table (Phase 1) + +| Unit | Score | Rationale | +|---|---:|---| +| `./src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt` | 5 | Composition root and the only public behavior class; carries Joran surface, hot path, three-stage lifecycle incl. the new restart support, metrics binding under a lock, `AppenderAttachable` - highest structural density and the unit where this session's lifecycle findings clustered | +| `./src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt` + `FallbackDispatcher.kt` (as a pair) | 5 | Two near-identical bounded-queue/worker/two-phase-close skeletons with cross-referencing KDoc ("mirrors", "canonical description lives there"); the pair diverged in one budget detail that the defect analysis had to fix | +| `./README.md`, `./docs/config/kafka-appender-config-guide.md`, `./docs/index.md`, `./docs/metrics/metrics-overview.md`, KDoc layer (as the self-description system) | 4 | Five hand-maintained surfaces carrying the same contracts and numbers; drift was the dominant failure mode in rounds 2-3 and recurred three times today | +| Metrics subsystem (`MetricsBindings.kt`, `MicrometerKafkaAppenderMetrics.kt`, `KafkaAppenderMetrics.kt`, `KafkaAppenderMetricsBinding.kt`) | 3 | ~30 % of the code for an optional feature; heavy but load-bearing per the round-2 decision (teardown lifecycle), re-verified; the binding's identity/state logic was simplified today | +| `./src/main/kotlin/eu/inqudium/tabellarium/ResilientMessageSender.kt` | 3 | Breaker/throttle/callback stage; today's `SendCallback` made captured state explicit - appropriate, no ceremony | +| Test architecture (`./src/test/kotlin/`, fixtures) | 3 | Excellent seams, but per-class private fixtures instead of shared support (four `RecordingAppender`s, ~17 producer doubles); the rule of three is met several times over | +| `./benchmarks/` | 3 | Regression instrument outside every build, reaching internals through compiler name mangling | +| `./src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt`, `ProducerPropertiesBuilder.kt`, `TopicClass.kt` | 2 | Per-class producer model with mandatory overrides; documented forces, YAGNI-disciplined future work (consolidation deferred) | +| `./src/main/kotlin/eu/inqudium/tabellarium/HalfOpenThrottle.kt`, `TopicRouter.kt`, `TopicTable.kt`, `TopicMappingConfig.kt`, `MessageEnricher.kt`, `KafkaProducerPropertiesParser.kt` | 1 | Small, pure, single-purpose; comment-heavy but each comment carries a rationale the code cannot show | +| `./pom.xml`, workflows, generators | 1 | Deliberate template investment with explicit purpose; the JDK floor is a documented trade-off (note in the baseline) | + +## 5. Findings + +### 🔴 Critical + +Nothing to report. The architecture does not obstruct the core task; the structure and the declared purpose (best-effort transport with visible loss and per-class producer policy) match. + +### 🟠 High + +Nothing to report. + +### 🟡 Medium + +- [x] 1. [Dispatcher pair - `./src/main/kotlin/eu/inqudium/tabellarium/SendDispatcher.kt:102` and `./src/main/kotlin/eu/inqudium/tabellarium/FallbackDispatcher.kt:95`; parallel-close loops `./src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt:843` and `./src/main/kotlin/eu/inqudium/tabellarium/ProducerRegistry.kt:116`] {Medium} {Confidence: high} {Under-Engineering / Consistency} The bounded-queue-worker skeleton and the two-phase close protocol exist as two hand-maintained copies that reference each other in prose instead of sharing code - and have already diverged once + - Actual structure: both dispatchers implement the same shape - `LinkedBlockingQueue` + one daemon worker, volatile `running`, `closeExecuted` CAS, `inFlight` compare-and-set ownership, uncaught-exception death handler that leaves the accepting state and drains, the reentry-guard mark, a graceful join then interrupt then bounded grace, and a post-close drain with accounting (168 vs. 144 code lines). Their KDoc cross-references the other eleven times (counting basis: grep for the sibling's name - 8 in `SendDispatcher.kt`, 3 in `FallbackDispatcher.kt`; phrases "mirrors", "same … protocol as", "the canonical description lives there"). On top, `KafkaAppender.closeSendDispatchersInParallel` and `ProducerRegistry.close` carry the same ~25-line "spawn daemon closers, join within a nanoTime deadline, restore the interrupt" loop twice. + - Solved problem / justifying force: the two dispatchers differ in what they *deliver* (send action versus `doAppend`) and in what a diverted item *becomes* (fallback with reason versus counted drop); those are real variations. Nothing justifies duplicating the skeleton around them: no ADR, no performance argument (the perf analysis measured the hand-off, not the class shape), and the differences are exactly the two injectable functions a shared skeleton would take. + - Cost: the copies drift. Today's defect analysis found the fallback dispatcher interrupting its drain after 200 ms while the send dispatcher used its full budget (finding 3 of `CODE_ANALYSIS-2026-09-07T19-09-00.md`) - a divergence that could not exist with one close protocol - and every remediation of the pair had to be made twice (death handlers in the August round, the reentry mark and the grace constant today). The prose cross-references are the symptom: they are what a maintainer reads *instead of* a shared type, and they cannot be checked by the compiler. Reach: two score-5 units plus two lifecycle loops. + - Simpler alternative: one `internal` bounded single-worker dispatcher parameterized by the delivery action and the divert/drop accounting (the two existing function-shaped seams), with the two-phase close and the death protocol written once; likewise one deadline-join helper for the two parallel-close loops. Not a framework - one class and one function, extracted from three-plus real cases. + - Reversibility: moderate and well-guarded - both dispatchers have exhaustive latch-anchored tests (23 methods) that would carry the extraction; the appender-level tests exercise the composed result. Worth doing at the next functional touch of either dispatcher rather than as a standalone rewrite. + - **Status:** Fixed in `698e2b3` (2026-09-07). New `BoundedWorkerDispatcher` holds the queue, worker, accepting state, in-flight ownership, death handler, two-phase close (drain budget, then interrupt plus the shared `INTERRUPT_GRACE_MS`) and reentry mark once; `SendDispatcher` (delivery = send action, rejection = claimed fallback diversion with the metric reason) and `FallbackDispatcher` (delivery = `doAppend`, rejection = counted drop) implement only the two abstract methods. New `ParallelClose.runWithin` replaces the two deadline-join loops in `KafkaAppender` and `ProducerRegistry`. The 23 dispatcher tests and the appender suite ran unchanged against the extraction; the prose cross-references are gone. + +- [x] 2. [Self-description system - `./docs/config/kafka-appender-config-guide.md:748-766` (defaults table, source "code"), `./README.md:284-298` (resilience numbers), `./docs/index.md:20-27`, `./docs/metrics/metrics-overview.md`, class-level KDoc of `FallbackDispatcher`/`SendDispatcher`/`ProducerRegistry`/`KafkaAppender`] {Medium} {Confidence: high} {Consistency} The same contracts and operational numbers are hand-maintained in up to five prose surfaces, so every structural change fans out into documentation edits - and they keep missing one + - Actual structure: the shutdown budgets, breaker thresholds, queue capacities and delivery semantics appear in the KDoc, the README, the configuration guide (twice: prose and a defaults table that states "code" as its source but is typed by hand), `docs/index.md` and the metrics overview. Round 3 already named "architecture self-description lags structural evolution" as a pattern (3 locations then); today the two defect passes recorded three new instances within hours of the changes (the 200 ms drain window surviving in the guide, the breaker-override promise contradicting ADR-0002 in three places, the `dispatched` semantics in the metrics overview). Coverage, test evidence and the badge, by contrast, are generated from builds and have never drifted. + - Solved problem / justifying force: rich rationale is a documented convention (CONTRIBUTING, ADR-0001) and a genuine strength - the KDoc layer *is* the design record. The force covers *having* the rationale; it does not cover *replicating* the same numbers and contracts by hand across five surfaces without a single source. + - Cost: measured, not presumed - four drift findings in two review days on ~2 100 lines of code, each costing a review cycle plus a PR to correct; readers of the guide and the README received contradictory numbers for a day. The cost scales with every future lifecycle or budget change (the units that change most). + - Simpler alternative: pick one canonical location per fact (KDoc for mechanism rationale, the guide's defaults table for numbers) and make the table *generated* from the constants - the project already owns the pattern (three small generators publish the test catalog, coverage and badge from build output); the README and `docs/index.md` then link rather than restate. Directional only; no tooling recommendation beyond what the repository already uses. + - Reversibility: cheap and incremental - documentation only; the generator is ~100 lines in the existing style; nothing in the code changes. + - **Status:** Fixed in `698e2b3` (2026-09-07). The guide's defaults quick reference is declared canonical and enforced by `DocumentationContractTest`, which compares every "code" row (queue capacities, drain budgets, interrupt grace, producer close timeout, breaker thresholds in both guide tables, probe gap, partitioning-key source) against the constants - mutation-checked: a changed number fails the build. The README resilience section links to the table instead of restating the numbers. Mechanism rationale stays in KDoc; the generator variant was not needed - a test is cheaper and equally binding. + +- [x] 3. [Restart capability - `./src/main/kotlin/eu/inqudium/tabellarium/KafkaAppender.kt:91-94` (contract), `:334-347` (fallback restart, guard re-arm), `:468-479` (breaker reset), `:801-806` (keep-attached stop), `./src/main/kotlin/eu/inqudium/tabellarium/KafkaAppenderMetricsBinding.kt:150-156` (state-based rebind)] {Medium} {Confidence: medium} {Over-Engineering / speculative feature} A `stop()` → `start()` restart is now a supported, tested capability of the appender - built in response to a defect finding, not to a consumer requirement + - Actual structure: since `1c8b048` and `5d849c8` (both today) the appender restarts an attached fallback appender, re-arms the one-shot error guard, resets every carried-over circuit breaker, keeps the fallback slot across `stop()`, exposes `isMeterRegistryBound` so the Spring binding can rebind the same instance, documents the contract in the class KDoc, and pins it with three tests. Logback's own `AsyncAppender` offers no such thing (its `stop()` detaches its appenders), and Logback's reconfiguration path never restarts an instance - it creates new ones. + - Solved problem / justifying force: the initial finding was real (a restart *silently* lost the fallback) - but its fix strategy named two directions, "refuse restart" or "make it symmetric", and the symmetric one was chosen without a force on record: no consumer, README or issue asks for programmatic restart, and the round-2 follow-up immediately found two more interactions the capability creates (breaker carry-over, binding identity). This is the YAGNI shape: a lifecycle branch that exists because it *can* be made correct, not because anyone needs it. + - Cost: four lifecycle sites plus one cross-class contract that every future stateful component must honor (each new per-appender resource now needs a "what happens on restart?" answer - the breakers already did), three tests, a paragraph of public contract that a consumer might start relying on. Reach: the composition root, the unit with the highest finding density in this session. + - Simpler alternative: refuse `start()` after `stop()` with an `addError` naming the reason (create a new instance), which removes the fallback-restart, breaker-reset and rebind branches and turns the KDoc paragraph into one sentence - *unless* a consumer requirement surfaces, in which case the current implementation is the right one and should be recorded as a deliberate API decision (a one-paragraph ADR, per ADR-0002's "conscious, documented API addition" rule). + - Reversibility: cheap now (pre-release, all sites from today), expensive later - once a consumer depends on restart it is API. The decision, not the code, is what this finding asks for. + - **Status:** Decided and fixed in `698e2b3` (2026-09-07) - restart is refused. `start()` after `stop()` fails with an `addError` naming ADR-0004 and the alternative (a new instance); the restart symmetry of the same day (fallback restart, breaker reset, error-guard re-arm) is removed, the three restart tests replaced by one refusal test. `docs/adr/ADR-0004-appender-instances-are-not-restartable.md` records the decision and the Logback-lifecycle rationale (a reconfiguration replaces instances; no consumer asks for same-instance restart); reinstating the capability is a follow-up ADR. The binding keeps deciding on the appender's bound state - simpler than the identity set regardless of restart. + +### 🟢 Low + +- [x] 4. [Test support - `./src/test/kotlin/eu/inqudium/tabellarium/KafkaAppenderTest.kt:81-134`, `ResilientMessageSenderTest.kt:36-64`, `SendDispatcherTest.kt:22-33`, `FallbackDispatcherTest.kt:26-39`, `ProducerRegistryTest.kt:26-50`] {Low} {Confidence: high} {Testability & Test Architecture (under-engineering)} Fixtures are copy-pasted per test class instead of shared: four private `RecordingAppender`s, ~17 producer-factory/double definitions across four files, six minimal encoders across three files (counting basis: grep for `class RecordingAppender`, `Producer by mock` / `: ProducerFactory {` / `ProducerFactory { _ ->`, `EncoderBase()`). The rule of three is exceeded several times; the round-2 thread-safety correction (`ThreadSafeListAppender`) landed as a fifth recorder next to the four private ones, and today's `SynchronousCallbackErrorProducer` is the third producer double that wraps a `MockProducer` for one behavior. Cost: each cross-cutting fixture fix is made N times or not at all; onboarding reads five recorders to learn one idea. Simpler alternative: one test-support file with the recorder, a minimal encoder and a small `MockProducer`-wrapping factory taking the behavior as a function - the seams already have that shape. Reversibility: trivial, test-only. + - **Status:** Fixed in `698e2b3` (2026-09-07). `TestSupport.kt` provides `RecordingAppender` (a started `ThreadSafeListAppender`, now `open`), `RecordingProducerFactory` (auto-complete flag plus a `wrap` seam for behavior doubles) and `MessageBytesEncoder`/`RecordingEncoder`/`ThrowingEncoder`; the four private recorders, four private factories and three private encoders are deleted. Test-specific doubles that model one behavior (blocking, self-logging, synchronous callback error) stay local by design. + +- [x] 5. [Benchmark module - `./benchmarks/pom.xml`, `./benchmarks/README.md:1-20`, `./benchmarks/src/main/java/eu/inqudium/tabellarium/bench/`] {Low} {Confidence: high} {Consistency / Dependency & Build Appropriateness} The project's only performance regression instrument is compiled by nothing: the module is "deliberately not part of the library's build", is not in the reactor, and no workflow builds it - while it reaches the library's `internal` seams through Kotlin's `$tabellarium` name mangling (`benchmarks/README.md`, "Conventions"), a compiler implementation detail that a module rename or an `internal` signature change breaks silently. The repository's own principle for the broker stage (round 3, finding 2: "runs only when a human remembers it" → own CI job "so it cannot rot silently") is not applied here. Cost: the benchmark inventory the BENCH_REPORT designates as "permanent regression asset" can already be broken today without anyone knowing. Simpler alternative: a compile-only step for the module in the CI workflow (no JMH run), or a `benchmark` profile inside the main module that shares its test seams without mangled names. Reversibility: cheap, additive. + - **Status:** Fixed in `698e2b3` (2026-09-07). The CI `build` job installs the library snapshot (tests, JaCoCo, ktlint, CycloneDX skipped) and compiles `benchmarks/` against it - compile only, no JMH run. Verified locally offline: the module compiles against the refactored dispatchers. The mangled-name access stays as documented; the compile step is what turns a signature change into a red build instead of silent rot. + +- [x] 6. [Operator documentation - `./README.md:783-793` ("Extension points": only the partitioning key), `./docs/config/kafka-appender-config-guide.md:766-768` ("code" defaults not exposed)] {Low} {Confidence: medium} {Boundaries & Responsibilities} The operator surface deliberately fixes breaker thresholds, probe gap, fallback capacity and all budgets in code (ADR-0002 discipline, appropriate), but the documentation states this only as a table footnote and, since today, a README sentence - there is no single place that lists *which* behaviors are intentionally not configurable and *why*, so each of the last two review days produced a question ("can operators tune the breaker?") that the docs answered in three different ways. Cost: small and recurring (review and support questions). Simpler alternative: one short "what is fixed and why" section in the guide, linked from the README extension-points section. Reversibility: trivial. + - **Status:** Fixed in `698e2b3` (2026-09-07). New guide section 13 "What is deliberately not configurable" lists breaker thresholds and probe gap, fallback capacity and drain budgets, the `max.block.ms` caps, the partitioning-key source, the serializers and the restart refusal, each with its reason; the README resilience and extension-points sections link to it. + +## 6. Systemic Patterns + +1. **Hand-rolled lifecycle skeletons instead of one shared one** - 4 occurrences (counting basis: the two dispatcher classes with the queue/worker/close protocol; the two nanoTime-deadline parallel-close loops in `KafkaAppender.kt:843` and `ProducerRegistry.kt:116`), held together by 11 prose cross-references between the dispatchers. Finding 1 carries it. The pattern statement: the project's discipline of documenting *why* two pieces of code must stay identical has substituted for making them the same code - and the defect history shows the substitution does not hold under change. + - **Status:** Fixed in `698e2b3` (2026-09-07) via finding 1 - one skeleton class and one parallel-close helper; occurrences 4 → 0. + +2. **Contracts and numbers replicated across hand-maintained surfaces** - 5 surfaces (KDoc, README, configuration guide prose + defaults table, `docs/index.md`, metrics overview; counting basis: full read of each against the code), with 3 drift instances recorded today on top of the 3 of round 3. Finding 2 carries it; finding 6 is one visible consequence. The contrast inside the same repository - generated coverage/test-evidence numbers that never drifted - is what makes it a pattern rather than a series of typos. + - **Status:** Fixed in `698e2b3` (2026-09-07) via findings 2 and 6 - one canonical, test-enforced table; README and KDoc link to it. The KDoc mechanism rationale remains by convention (ADR-0001) and is not a replication of numbers. + +**Positive counter-patterns (no findings, recorded for balance):** (a) every abstraction in the module has a demonstrated second case or a real seam - `KafkaAppenderMetrics` (no-op + Micrometer, required by optional-dependency gating), `ProducerFactory` (real client + `MockProducer`), the function-shaped dispatch/claim seams; no single-implementation interface exists (counting basis: the declaration inventory - 2 interfaces, both with two implementations); (b) YAGNI is applied where it is expensive to get wrong - producer-registry consolidation stays deferred pending a real deployment, the partitioning key stays unconfigurable pending an issue, the breaker registry stays internal per ADR-0002; (c) the remediation rounds keep *removing* rather than adding (synchronous test modes, all-open plugin, the binding's identity set today); (d) optional dependencies are gated uniformly (probe-then-bind for the Kafka Micrometer binder, `optional` for Spring/Micrometer/Logstash). + +--- + +*Pure analysis of commit `ffede8bcd89477d76e73d42035eab9c391a6570f`. This report is the only write operation; no code, configuration, or existing analysis document was modified. All improvement directions are deliberately described as strategy only.*