BatchProcessor

The Kafka positions a durable effect is answerable for: per partition, the highest offset seen plus one.

A type, and not a Vector[PendingWrite], because the wrong vector here is invisible and its consequence is silent. The writable records and the records the consumer is committing are different sets whenever anything dead-letters, and a signature that accepts the first will eventually be handed it — which is exactly the defect this type replaced: dead-lettered offsets never reached events.consumer_checkpoint, so the divergence gauge reported a permanent gap and restart?target=stored rewound onto records already on the DLQ. Building one takes ConsumerRecords, which the poison path has and the write path has, so the complete set is the natural thing to pass and the writable subset is the one you have to go out of your way to construct.

Plus one because that is Kafka's commit convention — the offset of the next record to read. Storing the last processed offset instead reads identically and is off by one at every seek, in the direction that reprocesses a record: safe, because the insert deduplicates, and still wrong.

How to turn a batch into the offsets it earned.

A small type rather than two loose parameters because the group id and the owner travel together everywhere and because it makes the absence of checkpointing one None instead of two — a processor built with a group id and no owner would otherwise be a state that compiles and means nothing.

Walks the cause chain looking for a SQLSTATE this build is willing to dead-letter a record for.

The chain matters: Magnum wraps, HikariCP wraps, and a BatchUpdateException's real reason is in getNextException rather than in getCause. Missing the state and treating a data error as transient is the benign direction — the stream restarts and eventually bisects to the same record — so the walk is deliberately conservative rather than clever.

SQLSTATE classes that mean "this row, not this database".

22 is a data exception (bad datetime, numeric overflow, invalid text representation); 23 is an integrity constraint violation, which on this schema is how cloud_event_specversion_ck, cloud_event_required_ck and "no partition of relation found for row" all surface. Both are deterministic properties of the row, so retrying or restarting can only reproduce them.

Everything else is treated as transient — including syntax and privilege errors (class 42), which are a build defect rather than a record defect and must page rather than quietly shovel the whole topic into the DLQ.