Stream processor

A consumer that reads a topic and adds up numbers is a morning's work. What takes the time is everything that happens when the input is not what you hoped: the same event delivered twice because a producer's acknowledgement timed out, a batch that arrives forty seconds after the window it belongs to, a payload that is not JSON at all.

Each of those has a wrong answer that looks entirely plausible. A duplicate inflates a subscriber's usage and can trip a threshold alert that should never have fired. A late batch gets counted in the wrong hour. A malformed record throws inside the consumer, kills the stream thread, and gets redelivered on restart to kill it again — a poison pill that stops the pipeline until somebody intervenes.

Records are consumed as strings and parsed explicitly, rather than being handed to a serde. A serde that throws does so inside the consumer, where there is no opportunity to route the offending record anywhere useful; the exception takes down the thread and the record is still there on restart.

Modelling the failure as data instead of an exception means a bad record becomes a value that can be branched on. Anything unparseable, or well-formed but missing the fields the pipeline needs, goes to a dead-letter topic with the original payload kept verbatim — a dead letter that has been helpfully cleaned up is useless, because the point is to replay exactly what arrived once the defect is fixed.

Kafka Streams emits an updated aggregate on every record by default. For a usage total that means a consumer sees a stream of partial answers it must know to discard, and the alerting branch fires on every record after the threshold rather than once per breach.

Suppressing until the window closes fixes both. The test that pins it down sends twelve records that collectively cross the limit and asserts exactly one alert — without suppression that is twelve alerts for one breach, which is how alerting systems get muted.

De-duplication keeps recently seen event ids in a state store. The first implementation keyed that store on the event id alone, which looked obviously correct and passed four of the five de-duplication tests.

The fifth sent two different subscribers an event carrying the same id. A Kafka partition holds many subscribers, and event ids are only unique within the system that issued them, so the second subscriber's event looked exactly like a redelivery of the first one's — and was silently discarded. Real usage, dropped, with nothing logged and no error anywhere.

The fix is one line: key the store on subscriber and event id together. The point is not the fix, it is that a plausible-looking implementation lost data in a way that no amount of staring at it would have revealed, and a test that took two minutes to write did.

The topology is a pure function from a configuration record to a Topology object. It reads no configuration, opens no connections, and imports nothing from Quarkus — the only framework-aware class in the repository is the one that supplies the config.

That makes the entire pipeline drivable through Kafka's TopologyTestDriver: real parsing, real state stores, real windowing and suppression, with event time under the test's control. The scenarios that are near-impossible to arrange against a live cluster — an event arriving after its grace expired, a producer retry, a poison payload — are ordinary unit tests that run in single-digit milliseconds.