> ## Content Index > Fetch the complete content index at: https://stalin-pereira.com/llms.txt > Use this file to discover other available public pages before exploring further. # Handling Duplicate Messages in Kafka Consumers - URL: https://stalin-pereira.com/handling-duplicate-messages-in-kafka-consumers/ - Published: 2025-12-24T15:41:00.000Z - Updated: 2026-10-16T19:23:20.000Z - Description: Kafka delivers at least once, and its exactly-once features stop at the edge of Kafka. Where duplicates come from, and how to build idempotent consumers with an inbox table, sane offset commits, and dead letter topics for poison pills. - Author: Stalin Pereira - Tags: Distributed Systems Your consumer reads a `PaymentReceived` event and credits the user's balance. Then it crashes before committing its offset. A few seconds later another consumer in the group picks up the partition, reads the same event, and credits the balance again. Nothing is broken. Kafka did exactly what it promised, and the user got paid twice. The fix isn't a broker setting. Kafka's default contract with consumers is at-least-once delivery, and its exactly-once features cover less ground than the name suggests. If your consumer has side effects outside Kafka, such as a database write, an email, or an API call, you have to make it safe to process the same message twice. Assume every event can arrive more than once and design for that. This article covers where duplicates come from, what Kafka's idempotence and transactions actually guarantee, how to build an idempotent consumer with an inbox table in the same database transaction as your business logic, and how to tell transient failures from poison pills so retries don't turn into a different kind of outage. ## Where duplicates come from Kafka tracks consumer progress with committed offsets. A consumer reads records, processes them, and commits the offset of the next record it wants. Anything that breaks the link between "processed" and "committed" causes redelivery. - **Crash between processing and committing.** The database write succeeded, the commit never happened. On restart, the consumer resumes from the last committed offset and processes the record again. - **Rebalances.** When consumers join or leave a group, partitions move. If a consumer loses a partition before committing what it processed, the new owner starts from the older committed offset. - **Slow processing.** If a consumer doesn't call `poll()` within `max.poll.interval.ms`, it's considered failed and removed from the group. Its partitions get reassigned while it may still be finishing the work, and its later commit is rejected. - **Producer-side duplicates.** An upstream service that times out and resends, an outbox relay that publishes a row then crashes before marking it sent, or a replay job: each puts a second copy of the event on the topic with a *different* offset. - **Operational replays.** Someone resets a consumer group's offsets to reprocess after a bug fix. Everything since that point comes through again. The last two points matter for design. Some duplicates are the same record redelivered (same partition and offset); others are the same event published twice (different offsets). A deduplication scheme keyed on offset handles the first kind and silently misses the second. ## What "exactly-once" in Kafka actually covers Kafka does have exactly-once machinery, and it's useful. It just doesn't solve this problem by itself. **Idempotent producers** (`enable.idempotence=true`, the default since Kafka 3.0) stop a producer's internal retries from writing the same record twice to a partition. That covers duplicates caused by the client library retrying a send. It doesn't cover your application calling `send()` twice, or a different producer instance publishing the same event. **Transactions** let a producer write to several partitions and commit consumer offsets atomically. Combined with consumers using `isolation.level=read_committed`, this gives exactly-once semantics for read-process-write pipelines where both the input and the output are Kafka topics. Kafka Streams exposes this as `processing.guarantee=exactly_once_v2`. The boundary is Kafka itself. Once your consumer writes to PostgreSQL, charges a card or sends an email, Kafka's transaction can't include that side effect. The broker has no way to roll back your SQL `UPDATE` if the offset commit fails. For those consumers, exactly-once delivery isn't available. What you can build is exactly-once *effect*: at-least-once delivery plus idempotent processing. ## The idempotent consumer The pattern has three parts: 1. Every event carries a stable, unique ID assigned by the producer when the event is created, not when it's sent. Retries and replays carry the same ID. 2. The consumer records each processed ID in an inbox table with a unique constraint. 3. Recording the ID and applying the business change happen in **one database transaction**. The third point is what makes it work. If you check the inbox, do the work, and then insert the ID in a separate step, a crash in between leaves you with the work done and no record of it. Doing both in one transaction means there are only two possible outcomes: both happened or neither did. The schema: ```sql CREATE TABLE processed_events ( consumer_name text NOT NULL, event_id uuid NOT NULL, processed_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (consumer_name, event_id) ); CREATE INDEX processed_events_processed_at_idx ON processed_events (processed_at); ``` Including `consumer_name` in the key lets several independent consumers share the table without stepping on each other. The billing consumer and the notifications consumer both need to process the same event once each. The processing logic, using plain JDBC against PostgreSQL: ```java import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.UUID; import javax.sql.DataSource; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.header.Header; public final class PaymentReceivedHandler { private static final String CONSUMER = "billing-balance-updater"; private final DataSource dataSource; public PaymentReceivedHandler(DataSource dataSource) { this.dataSource = dataSource; } /** Applies the event at most once. Safe to call repeatedly with the same record. */ public void handle(ConsumerRecord record) throws SQLException { UUID eventId = eventId(record); PaymentReceived event = record.value(); try (Connection conn = dataSource.getConnection()) { conn.setAutoCommit(false); try { if (!markProcessed(conn, eventId)) { conn.rollback(); // already applied, nothing to do return; } try (PreparedStatement ps = conn.prepareStatement( "UPDATE accounts SET balance = balance + ? WHERE id = ?")) { ps.setLong(1, event.amountMinor()); ps.setLong(2, event.accountId()); ps.executeUpdate(); } conn.commit(); } catch (SQLException e) { conn.rollback(); throw e; } } } private static boolean markProcessed(Connection conn, UUID eventId) throws SQLException { try (PreparedStatement ps = conn.prepareStatement( "INSERT INTO processed_events (consumer_name, event_id) VALUES (?, ?) " + "ON CONFLICT DO NOTHING")) { ps.setString(1, CONSUMER); ps.setObject(2, eventId); return ps.executeUpdate() == 1; } } private static UUID eventId(ConsumerRecord record) { Header h = record.headers().lastHeader("event-id"); if (h == null) { throw new InvalidEventException("missing event-id header"); } return UUID.fromString(new String(h.value(), StandardCharsets.UTF_8)); } } ``` (`PaymentReceived` is your event type and `InvalidEventException` is a plain `RuntimeException` subclass; both are omitted for brevity.) Using `ON CONFLICT DO NOTHING` rather than catching a duplicate key exception keeps the transaction usable. In PostgreSQL, a failed statement aborts the whole transaction, so the common "insert, catch the duplicate key error, ignore it" approach needs a savepoint or a rollback to work correctly. Two concurrent deliveries of the same event, which can briefly happen during a rebalance, are also handled. The second insert blocks on the first transaction's uncommitted row, then sees the conflict once the first commits and skips. ## The poll loop and offset commits With idempotent processing in place, the commit strategy can stay simple: disable auto-commit, process a batch, then commit. If the consumer dies mid-batch, the records it already applied will be redelivered and skipped. ```java Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing-balance-updater"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, PaymentReceivedDeserializer.class); try (KafkaConsumer consumer = new KafkaConsumer<>(props)) { consumer.subscribe(List.of("payments.received")); while (running) { ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord record : records) { handler.handle(record); } consumer.commitSync(); } } ``` Keep `max.poll.records` small enough that a batch finishes well within `max.poll.interval.ms`. If processing routinely runs long, you'll trigger rebalances, and every rebalance means more redelivery. Idempotency makes that safe, but it's still wasted work. This loop has no error handling yet, which is the next problem. ## Retries, poison pills and dead letter topics Not every failure should be retried, and the two kinds need opposite handling: - **Transient errors**, such as the database being unavailable, a lock timeout, or a downstream 503, will probably succeed if you wait. You *want* redelivery here: don't commit, back off, try again. - **Poison pills**, such as malformed JSON, a missing required field, or an event referencing an account that doesn't exist, will fail the same way every time. Retrying them blocks the partition indefinitely, because Kafka delivers in order within a partition and the consumer can't move past the bad record. Poison pills belong on a dead letter topic (DLT): publish the record there along with the error details, commit past it, and keep going. Someone or something inspects the DLT later. If you use Spring for Apache Kafka, its `DefaultErrorHandler` does most of this. It retries with a back-off, then hands the record to a recoverer, and you mark which exceptions should skip the retries: ```java @Bean DefaultErrorHandler errorHandler(KafkaTemplate template) { var recoverer = new DeadLetterPublishingRecoverer(template, (record, ex) -> new TopicPartition(record.topic() + ".DLT", -1)); var handler = new DefaultErrorHandler(recoverer, new ExponentialBackOff(500L, 2.0)); handler.addNotRetryableExceptions(InvalidEventException.class); return handler; } ``` The negative partition lets Kafka choose the DLT partition, so the DLT doesn't need the same partition count as the source topic. In real use, give the `ExponentialBackOff` a maximum elapsed time; otherwise transient errors retry for a very long time. Deserialization failures need an `ErrorHandlingDeserializer` wrapping your real deserializer, so they reach the error handler instead of failing inside `poll()`. Two cautions. First, a DLT that nobody watches is just a slower way of dropping data. Alert on it and have a replay path. Second, events sent to the DLT and later replayed keep their event IDs, so the inbox table protects you there too, as long as replay goes through the same consumer logic. ## Practical details that bite later **Pick the right ID.** Use an ID the producer assigns once per business event, in a header or the payload. Don't generate it at send time inside a retry loop, and don't rely on `topic-partition-offset` unless you're sure duplicates can only come from redelivery. Where the event has a natural key, such as a payment ID for `PaymentReceived`, that's often the best ID of all. **Plan the table's growth.** The inbox table grows with every event. Delete rows older than the longest replay window you'll ever need. If someone replays a topic from further back than your retention, deduplication won't catch it, so make sure retention and replay policy are decided together. **Prefer naturally idempotent writes where you can.** "Set order status to SHIPPED" or "upsert the customer's latest address" can be repeated safely without an inbox table. Increments, appends and external calls can't. Conditional updates with a version column (`UPDATE ... WHERE id = ? AND version = ?`) are another option when events carry versions. **External side effects need their own idempotency.** The inbox table only protects writes inside the same database. If the consumer calls a payment provider or sends an email, the call can't join your transaction. Pass a deterministic idempotency key derived from the event ID to the downstream API, which is the same technique described in [Designing Idempotency Keys for Payment APIs](https://stalin-pereira.com/designing-idempotency-keys-for-payment-apis/). Or record the intent in your database and have a separate worker perform the call. **Storing offsets in your database is an alternative.** You can write the consumed offset into the same transaction as the business change, then on partition assignment `seek()` to the stored offset using a `ConsumerRebalanceListener`. That handles redelivery cleanly but does nothing about events published twice upstream, so an event-ID inbox is usually the more robust choice. ## When to use which approach - **Kafka in, Kafka out, no external side effects:** use Kafka transactions or Kafka Streams with `exactly_once_v2`. You get real exactly-once processing within Kafka. - **Kafka in, database write out:** use an inbox table in the same transaction as the business write. This is the default for most service consumers. - **The write is a pure upsert or state overwrite:** skip the inbox table; the operation is already idempotent. - **Kafka in, external API call out:** use a downstream idempotency key derived from the event ID, or an outbox so the call happens from a worker you control. - **Analytics, metrics or logging where an occasional double count is acceptable:** you may not need any of this. Say so explicitly in the design rather than discovering it later. Before shipping a consumer, test it directly: deliver the same record twice, kill the process between the database commit and the offset commit, and publish the same event ID at two different offsets. If the balance is right after all three, the consumer is ready.