Recovering Kafka Notification Consumers: CAS, Offsets, and External Side Effects

A Kafka record can be delivered again even when its first attempt sent the notification successfully. The consumer may stop after the external provider accepts the request but before the database records completion or Kafka stores the offset.

I used compare-and-set to claim each notification and separate error paths for invalid data and infrastructure failures. The difficult part was deciding when to restore the database state and when Kafka acknowledgment was safe.

The upstream API and publisher are covered in the architecture overview. At this boundary, the consumer receives a Kafka record and must decide whether to perform the notification, retry it, or record a permanent failure.

Claim the notification with an explicit transition

The project used the shared outbox state to distinguish publication from consumption:

SENT (2)
  -> tryMarkConsuming (CAS) -> CONSUMING (5)
       -> processing succeeds -> CONSUMED (4)
       -> DataError -> FAILED (3), then acknowledge
       -> SystemError -> SENT (2), then stop without acknowledging

tryMarkConsuming was the gate before delivery. A simplified SQL representation of that compare-and-set is:

UPDATE outbox
SET send_status = 5
WHERE id = :outbox_id
  AND send_status = 2;

The SQL illustrates the recorded state transition; it is not a complete repository implementation. Check the affected-row count before calling the provider. Two consumers cannot both change the same row from SENT to CONSUMING through that conditional update. PostgreSQL's concurrent update behavior explains the predicate recheck after a conflicting update.

That gate protects a state transition, not the entire external delivery. A zero-row result needs interpretation: the row may already be consumed, still be in progress, or not yet have been marked sent by the publisher. These cases should not all be acknowledged as completed duplicates.

Classify the failure before deciding whether to acknowledge

The implementation treated malformed JSON, missing fields, and invalid email data as DataErrorException. Retrying the same payload would not repair those problems. Failures involving the provider, object storage, or database were classified as SystemErrorException, preserving the obligation for another attempt.

The error branches followed this shape:

catch (e: DataErrorException) {
    markPort.markPermanentFailure(outboxId, e.message)
    ack.acknowledge()
} catch (e: SystemErrorException) {
    markPort.rollbackConsumingToSent(
        outboxId,
        nextAttemptDelaySec = retryDelaySeconds
    )
    SpringApplication.exit(ctx)
    exitProcess(1)
}

The first branch records permanent failure before acknowledging. The second restores SENT and exits without acknowledging, allowing replay after restart.

Here, rollbackConsumingToSent means a compensating state update. It cannot undo a notification the external provider already accepted. If recording failure or restoring state itself fails, the desired database outcome has not been established merely because the exception was caught.

Notification payloads also shaped the worker boundary

The project sent encrypted token material through the outbox instead of placing a raw verification token in Kafka. The API retained a token hash for later verification; the notification consumer decrypted the delivery payload and injected the resulting link or code into the email template. The payload included a key identifier and format version, and the encryption context included the token purpose.

That gave the consumer a concrete responsibility beyond forwarding a record: interpret the notification purpose, prepare the template model, and call the email provider. A bad payload and an unavailable dependency therefore needed different failure policies. A successful send also remained separate from the recipient later using the verification link.

Consumer recovery must reach beyond compare-and-set

The implemented error branches deliberately favored continuing past invalid data and retrying after infrastructure failure.

That policy has a useful intent: one permanently invalid record should not repeatedly block valid work, while a temporary outage should not erase the obligation. Its correctness depends on details outside the catch block.

First, persist quarantine successfully before acknowledging a bad record. If the database is unavailable, the failure cannot be considered recorded. For a record that cannot even be decoded, preserve its topic, partition, and offset through an appropriate deserialization recovery path; the application listener may never receive it.

Second, distinguish why a consumer claim failed:

• Already completed: a duplicate can be acknowledged.
• Currently owned by another live attempt: defer or retry according to a defined policy.
• Not yet eligible: do not mistake delay for completion.
• Missing or inconsistent state: investigate or recover; do not silently skip it.

Third, a caught infrastructure exception is only one interruption path. A forced termination after claiming skips the cleanup catch block entirely. A persisted consuming state therefore needs a lease, a recovery sweep, or another explicit way to become eligible again. “Restart the process” by itself cannot repair an orphaned database state.

Finally, database CAS cannot make an external side effect exactly once:

provider accepts the notification
process stops before recording completion
record is delivered again

A unique event ID plus a transactional consumer receipt works when the receipt and business effect live in the same database transaction. For an external notification API, use a stable provider idempotency key if that API supports it. Otherwise, define how ambiguous outcomes are reconciled and acknowledge the possibility of duplicates. Recording completion before calling the provider merely trades duplicates for potential loss.

These limits explain where the implemented CAS and compensating update stop providing protection.

Acknowledgment is a configured behavior

Calling acknowledge() is not enough to describe offset durability. Spring Kafka distinguishes manual acknowledgment modes, and the timing of commits depends on the listener container. The container reference documents those differences.

Also, not acknowledging one record does not make it safe to acknowledge later records from the same partition indiscriminately: a committed offset represents progress through that partition. Keep processing and recovery consistent with the chosen acknowledgment mode and concurrency model.

The original process-exit strategy is simple to understand, but can interrupt unrelated partitions and turn a prolonged dependency outage into a restart loop. A configured container error handler with bounded retries, pausing, and durable recovery can provide finer control. Spring Kafka's error handling reference describes these mechanisms. Changing to them requires tests of the actual configuration; it is not a cosmetic replacement of exitProcess.

Turn the policy into a failure matrix

Use this matrix to validate the policy. Rows involving lease expiration or generations apply only if those extensions are added.

Use a real Kafka broker and PostgreSQL database, plus a controllable provider stub. The stub needs two separate observations: whether it accepted a request and whether the client received a response. Otherwise the most important ambiguous outcome cannot be reproduced.

Read committed consumer-group offsets, not just application acknowledgment logs. Include a rebalance during processing and a forced process termination: graceful shutdown alone can hide orphaned claims.

If a retry eligibility timestamp is stored in PostgreSQL, verify the Kafka redelivery path actually respects it. Kafka does not automatically read that timestamp. A paused partition, container retry, or separate durable retry queue must implement the delay. Each choice affects ordering and the latency of unrelated records.

Extension: separate delivery receipts and recover abandoned claims

The shared-state design can be extended with separate consumer receipts and lease generations. A receipt keyed by event ID and consumer purpose can separate delivery completion from publisher bookkeeping. A lease can make an abandoned claim recoverable, while a generation guards against stale workers finalizing another attempt.

Receipts provide visible recovery state but add database writes and a retention policy. Keep deduplication evidence at least as long as the supported replay horizon. If old Kafka records can be replayed after receipts are deleted, the same notification can become eligible again.

Per-consumer receipts also introduce a shared database dependency. They fit when notification volume and availability requirements permit that coordination and when an operator needs to investigate individual delivery outcomes. They are less attractive when database round trips dominate the workload or consumers must operate independently of that database.

Stopping the whole process on infrastructure failure, as the project implemented, is a coarse recovery mechanism. Pausing or retrying individual work can reduce disruption, but introduces scheduling and ordering choices. Neither strategy removes the need to recover persisted claims after a hard crash.

Most importantly, choose the product's ambiguity policy deliberately. A harmless duplicate notification and a non-repeatable external action have different requirements. Without provider-side idempotency or a way to query the result, neither a CAS flag nor a Kafka transaction can prove that an external action happened exactly once. Kafka's delivery semantics explain the boundary with external systems.

The consumer's completion test is a durable outcome, not merely a successful claim. For the separate problem of getting a committed event into Kafka, see recoverable PostgreSQL outbox publishers.