BatchProcessor

Turns one groupedWithin batch into a durable effect, and returns the offsets that effect earned.

This class is the at-least-once guarantee. It returns Committables, and it returns them only on the paths where the batch is genuinely accounted for — every record either written to PostgreSQL, or published to the DLQ. When neither is true the returned Future fails, the stream fails with it, and the offsets are never handed to the committer. ConsumerStream then puts Committer.flow strictly downstream of this stage, so an offset is only ever a receipt for work that already happened.

Poison isolation is a bisection, not a retry loop. ADR §4.3 requires that one bad record cannot wedge the stream. The naïve fix — dead-letter the whole batch on failure — discards up to batchSize perfectly good events for one malformed one. So a failed batch is retried whole (a database blip is not a data problem), then split in half and each half written independently, recursively, until the failure is attributed to a single record. That record is dead-lettered and the batch proceeds. log₂(500) ≈ 9 extra round trips is a cheap price for never losing a good event and never stalling.

The classification in BatchProcessor.isDataError is what keeps the bisection safe. Without it, a database that is merely down would bisect to singletons and dead-letter the entire batch — converting a recoverable outage into permanent data loss, which is strictly worse than stalling. Only a failure the database will deterministically repeat (SQLSTATE class 22, data exception; class 23, integrity constraint) is allowed to dead-letter a record. Everything else is rethrown so the RestartSource backs off and tries again with the offsets uncommitted.

The externalised checkpoint and the Kafka commit always describe the same position, and that is the invariant BatchProcessor.Accounted exists to hold. process returns a Committable for every record in the batch, decoded or not, so the checkpoint has to account for every one of them too. Deriving it from the writable records alone — which is what a Vector[PendingWrite] invites — silently drops the offsets of dead letters, and the two consequences are both quiet: consume.checkpoint.divergence, whose only healthy value is zero, reports a gap that means nothing and hides the one that does; and restart?target=stored rewinds the group onto records it has already dead-lettered, re-consuming and re-dead-lettering them on every recovery. The offsets are therefore computed once, in process, from the whole batch, and insert is the single place they can reach the database.

No span is opened around the write. A batch aggregates records from many unrelated traces, so it cannot be a child of any one of them, and a span with batchSize links is not something any backend renders usefully. The per-record CONSUMER span in RecordDecoder is where trace continuation lives; batch health is a metric (ConsumerMetrics.batchWrite), which is the shape that question actually has.