Fast Kafka consumers are not automatically safe Kafka consumers. A Java service can keep up with partitions, restart cleanly, and still leave the business in a state that is hard to explain after a retry, rebalance, offset rewind, or partial outage.
For business-critical event streams such as inventory, usage metering, billing, fulfillment, fraud detection, or security analytics, the real test is not only “did the consumer process the message?” The harder test is: can the team replay a bounded slice of history and trust the state produced by that replay?
This article focuses on one practical layer of that problem: how to design a Java Kafka consumer so duplicate delivery, operational replay, and recovery workflows are safe by construction. The examples use synthetic inventory events, but the pattern applies to many high-throughput enterprise systems.
Duplicate Delivery Is a Normal Case
Kafka consumers should be designed with the assumption that the same business event may be observed more than once. That can happen because of retries, consumer restarts, rebalance timing, offset rewind, replay jobs, or downstream recovery drills.
A simple consumer loop usually looks like this:
ConsumerRecords<String, InventoryEvent> records =
consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, InventoryEvent> record : records) {
inventoryService.apply(record.value());
}
consumer.commitSync();
This is fine as a sketch, but it hides the most important question in the system: where does the application record that a business event has been accepted?
If the aggregate table is updated, a downstream message is published, and the Kafka offset is committed without a durable acceptance record, a replay can become guesswork. The offset tells you how far the consumer moved. It does not necessarily tell you which business facts were safely applied.
Replay-safe consumers make acceptance explicit.
Define the Consumer Contract
Before writing the consumer code, define a small contract for the stream:
Consumer contract:
authoritative history: accepted inventory transaction table
idempotency key: eventId
ordering boundary: sellerId + sku
business transaction: accept event and update aggregate together
offset rule: commit after business transaction
replay scope: sku, seller, time window, offset range, or partition
evidence: accepted count, duplicate count, changed rows, failed checks
This is not a framework. It is a way to make recovery assumptions visible. If a team cannot name the idempotency key, ordering boundary, replay scope, and reconciliation checks, the consumer is probably relying on tribal knowledge.
Start With a Stable Event ID
Every business event should carry a stable idempotency key. The key should be generated before the event reaches the consumer and remain stable across producer retries, topic replays, and recovery jobs.
{
"eventId": "evt-8f11a",
"sellerId": "seller-42",
"sku": "SKU-12345",
"deltaQuantity": 100,
"eventTime": "2026-06-19T18:23:11Z"
}
The key is not just for logging. It is part of the data contract. The consumer must be able to observe this event five times and apply the inventory delta once.
In Java, the event can be represented with a compact immutable type:
public record InventoryEvent(
String eventId,
String sellerId,
String sku,
long deltaQuantity,
Instant eventTime
) {
public String orderingKey() {
return sellerId + ":" + sku;
}
}
The `orderingKey` is not a convenience method. It documents the business entity whose state must be rebuilt deterministically.
Store Event Acceptance Before Changing State
A common approach is to insert the event into an accepted-transaction table before changing aggregate state.
CREATE TABLE inventory_transaction (
event_id VARCHAR(128) PRIMARY KEY,
seller_id VARCHAR(64) NOT NULL,
sku VARCHAR(64) NOT NULL,
delta_quantity BIGINT NOT NULL,
event_time TIMESTAMP NOT NULL,
accepted_at TIMESTAMP NOT NULL
);
CREATE TABLE inventory_stock (
seller_id VARCHAR(64) NOT NULL,
sku VARCHAR(64) NOT NULL,
quantity BIGINT NOT NULL,
updated_at TIMESTAMP NOT NULL,
PRIMARY KEY (seller_id, sku)
);
The accepted transaction table becomes the recovery anchor. If a projection, cache, materialized view, or downstream topic drifts, the team has a durable history from which to rebuild.
Make the Business Write Idempotent
The application service should treat duplicate detection as part of the write path, not as a separate cleanup job.
public ApplyResult applyInTransaction(InventoryEvent event) {
boolean accepted = transactionRepository.tryInsertAcceptedEvent(event);
if (!accepted) {
metrics.increment("inventory.consumer.duplicate_suppressed");
return ApplyResult.duplicate(event.eventId());
}
stockRepository.incrementStock(
event.sellerId(),
event.sku(),
event.deltaQuantity()
);
metrics.increment("inventory.consumer.accepted");
return ApplyResult.accepted(event.eventId());
}
The important part is the transaction boundary. The accepted-event insert and the aggregate update must commit together. If they are split across two database transactions, the system can record that an event was accepted without applying it, or apply it without recording the acceptance fact.
For PostgreSQL, the repository can use `ON CONFLICT DO NOTHING`:
INSERT INTO inventory_transaction (
event_id,
seller_id,
sku,
delta_quantity,
event_time,
accepted_at
) VALUES (
:event_id,
:seller_id,
:sku,
:delta_quantity,
:event_time,
now()
)
ON CONFLICT (event_id) DO NOTHING;
The repository should return whether one row was inserted. If the insert count is zero, the event already became a business fact and the consumer should not apply the delta again.
public boolean tryInsertAcceptedEvent(InventoryEvent event) {
int rows = jdbcTemplate.update(INSERT_ACCEPTED_EVENT_SQL, params(event));
return rows == 1;
}
That one boolean is doing a lot of work. It turns duplicate Kafka delivery into a safe no-op at the business layer.
Commit Offsets After the Business Transaction
Kafka offsets and business state are different forms of progress. Treating the offset as the only truth is a common source of fragile recovery behavior.
If the consumer commits the Kafka offset before the database transaction completes, a crash can lose the business update. Kafka believes the message has been consumed, but the database never accepted it.
If the consumer commits the offset after the database transaction completes, a crash can cause the same Kafka message to be delivered again. That is usually safer because the idempotency key turns the duplicate into a no-op.
while (running.get()) {
ConsumerRecords<String, InventoryEvent> records =
consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, InventoryEvent> record : records) {
ApplyResult result =
inventoryService.applyInTransaction(record.value());
auditLogger.record(
record.topic(),
record.partition(),
record.offset(),
result
);
}
consumer.commitSync();
}
This pattern does not make offset management unimportant. It makes the offset a delivery checkpoint, not the only recovery record.
Handle Batches Without Hiding Partial Progress
Batch processing improves throughput, but it can hide partial progress if every record is treated as one giant unit. For replay-safe consumers, a batch should still produce per-event acceptance results.
public BatchResult applyBatch(List<InventoryEvent> events) {
BatchResult result = new BatchResult();
for (InventoryEvent event : events) {
try {
ApplyResult applied = applyInTransaction(event);
result.record(applied);
} catch (RetryableDataAccessException retryable) {
result.retry(event, retryable);
} catch (InvalidInventoryEventException invalid) {
result.reject(event, invalid);
}
}
return result;
}
Do not collapse accepted, duplicate, retryable, and rejected outcomes into a single “failed batch” metric. During replay, those categories matter.
Keep the Ordering Boundary Explicit
Kafka preserves order within a partition. It does not know which business entity requires ordered updates. That is the application’s responsibility.
For inventory, the ordering boundary might be `sellerId + sku`. For billing, it might be `tenantId + accountId`. For security analytics, it might be `userId`, `deviceId`, or `sessionId`.
The producer should key related events consistently:
public ProducerRecord<String, InventoryEvent> toRecord(InventoryEvent event) {
return new ProducerRecord<>(
"inventory.events.v1",
event.orderingKey(),
event
);
}
If related events are round-robined across partitions, the consumer may observe a sequence that no reconciliation query can cleanly explain. The practical rule is simple: choose the Kafka key from the entity whose state must be rebuilt deterministically.
Separate Retry From Rejection
A replay-safe consumer should distinguish retryable infrastructure problems from permanent business rejection.
try {
ApplyResult result = inventoryService.applyInTransaction(event);
metrics.increment("inventory.consumer.result", "status", result.status());
} catch (TransientDatabaseException transientFailure) {
throw transientFailure; // retry through consumer policy
} catch (InvalidInventoryEventException invalidEvent) {
rejectionRepository.record(recordMetadata, event, invalidEvent);
metrics.increment("inventory.consumer.rejected");
}
A malformed event, missing required field, unsupported schema version, or invalid business operation should be recorded as rejected. A database timeout, temporary lock conflict, or unavailable dependency should usually be retried.
Dead-letter queues are useful, but they are triage tools, not recovery plans. A DLQ can tell you which messages failed. It does not automatically prove that derived state is now correct.
Make Replay a First-Class Path
Replay should not mean “reset offsets and hope.” A replay workflow should name the source of history, the scope of replay, the projection being rebuilt, and the checks that prove the rebuilt state is trustworthy.
public record ReplayRequest(
String flow,
String sellerId,
String sku,
Instant fromEventTime,
Instant toEventTime,
boolean dryRun
) {}
The replay path should read authoritative history, rebuild deterministic state, run checks, and only publish if those checks pass.
Return Evidence From Replay
The strongest replay workflows produce evidence, not just new state.
public RecoveryEvidence replay(ReplayRequest request) {
List<InventoryEvent> history =
historyRepository.loadAcceptedEvents(request);
InventoryProjection before =
projectionRepository.currentProjection(request.sellerId(), request.sku());
InventoryProjection rebuilt =
projectionEngine.rebuild(history);
ReconciliationResult reconciliation =
reconciliationService.compare(request, rebuilt);
if (!request.dryRun() && reconciliation.passed()) {
projectionRepository.publish(rebuilt);
}
RecoveryEvidence evidence = RecoveryEvidence.builder()
.flow(request.flow())
.sellerId(request.sellerId())
.sku(request.sku())
.eventsRead(history.size())
.projectionChanged(!rebuilt.equals(before))
.reconciliationPassed(reconciliation.passed())
.failedChecks(reconciliation.failedChecks())
.published(!request.dryRun() && reconciliation.passed())
.build();
evidenceRepository.save(evidence);
return evidence;
}
This gives operators something concrete to inspect after recovery. Instead of asking “did the replay run?”, they can ask:
- Which entity and time window were replayed?
- How many events were read?
- Were duplicates suppressed?
- Did the rebuilt projection match source-of-truth history?
- Which checks failed?
- Was anything republished downstream?
That is the difference between restartability and confidence.
Add Recovery Metrics
Consumer lag matters, but lag alone does not prove correctness. A pipeline can be caught up and still be wrong.
Track ordinary health:
- consumer lag
- poll duration
- commit failures
- retry count
- rejection count
Track recovery confidence:
- duplicate suppression count
- replay duration
- replay scope size
- projection rows changed
- reconciliation failures
- evidence reports produced
Example metric names:
inventory_consumer_events_total{status="accepted|duplicate|rejected"}
inventory_consumer_commit_failures_total
inventory_consumer_duplicate_suppressed_total
inventory_replay_duration_seconds
inventory_replay_events_read_total
inventory_replay_projection_changed_total
inventory_reconciliation_failures_total
These metrics help operators tell the difference between a consumer that is merely running and a consumer whose state can be trusted.
Tests Worth Automating
Replay-safe consumers need tests beyond the happy path.
@Test
void duplicateEventDoesNotApplyDeltaTwice() {
InventoryEvent event = new InventoryEvent(
"evt-1",
"seller-42",
"SKU-12345",
10,
Instant.parse("2026-06-19T18:23:11Z")
);
inventoryService.applyInTransaction(event);
inventoryService.applyInTransaction(event);
assertThat(stockRepository.find("seller-42", "SKU-12345").quantity())
.isEqualTo(10);
assertThat(transactionRepository.countAccepted("evt-1"))
.isEqualTo(1);
}
Also test late events, schema-compatible changes, partition-key mistakes, offset rewinds, retryable database failures, and replay dry runs.
@Test
void replayDryRunReturnsEvidenceWithoutPublishing() {
ReplayRequest request = new ReplayRequest(
"inventory-availability",
"seller-42",
"SKU-12345",
Instant.parse("2026-06-19T18:00:00Z"),
Instant.parse("2026-06-19T19:00:00Z"),
true
);
RecoveryEvidence evidence = replayService.replay(request);
assertThat(evidence.reconciliationPassed()).isTrue();
assertThat(evidence.published()).isFalse();
assertThat(evidence.eventsRead()).isGreaterThan(0);
}
The point is not to test Kafka itself. The point is to test the consumer’s recovery contract.
Production Checklist
Before calling a Java Kafka consumer replay-safe, confirm:
- Every business event has a stable idempotency key.
- Business acceptance is recorded durably.
- Event acceptance and aggregate updates share one database transaction.
- Kafka offsets are committed after the business transaction.
- The partition key matches the business ordering boundary.
- Duplicate, rejected, retryable, and accepted outcomes are measured separately.
- Replay can be bounded by entity, time, offset, or partition.
- Reconciliation checks are executable.
- Replay can run in dry-run mode.
- Replay produces a structured evidence report.
- Operators can find the recovery runbook from the alert.
Conclusion
Replay-safe consumers are not built by Kafka configuration alone. They are built through disciplined Java design choices: stable event IDs, transactional event acceptance, idempotent aggregate updates, explicit ordering boundaries, safe offset commits, bounded replay, and recovery evidence.
When those choices line up, replay becomes more than a debugging trick. It becomes a controlled way to rebuild state and explain why that state should be trusted.
References
- Apache Kafka Documentation: https://kafka.apache.org/documentation/
- Spring Framework Declarative Transaction Management: https://docs.spring.io/spring-framework/reference/data-access/transaction/declarative.html
- PostgreSQL INSERT and ON CONFLICT documentation: https://www.postgresql.org/docs/current/sql-insert.html
