Handling Duplicate Message #

When we configure an Apache Kafka consumer using the At-Least-Once processing guarantee (to ensure no business message is ever lost), the unavoidable logical consequence is the appearance of message duplication. Unstable internet networks, failed offset commit writes mid-transaction, or dynamic rebalance processes are common events in distributed server clusters. Handing message processing order entirely to the downstream database without special protection will damage our business data consistency (like debiting a customer’s balance multiple times). Therefore, we must build an Idempotent Consumer mechanism to filter duplicate messages before triggering sensitive business logic.

Why Do Duplicate Messages Happen in Kafka? #

Before designing a solution, we must precisely understand how message duplication sneaks into our system. There are two classic scenarios that most often happen in production:

Scenario 1: Offset Commit Delivery Failure (Network Timeout) #

  1. A producer sends a message to the Kafka broker with the unique transaction key TX-999 (Offset 500).
  2. Our consumer successfully fetches that message through the .poll() function call.
  3. The consumer processes the transaction, deducts the balance, and writes the successful transaction record to our business database.
  4. Problem: When the consumer sends the commitSync(Offset 500) signal back to the broker, the network connection drops temporarily or the coordinator broker experiences a pause. The broker never receives that commit.
  5. After the connection recovers, the consumer (or a new post-rebalance consumer) finds the last committed offset still at 499.
  6. The consumer is pulled back to re-read Offset 500 (TX-999). If the database has no protective filter, the customer’s balance is deducted a second time.

Scenario 2: Rebalance Mid-Batch Processing #

  1. The consumer calls .poll() and receives 50 messages at once (Offsets 1000 to 1049).
  2. The consumer starts processing messages one by one synchronously on the main thread.
  3. After successfully processing 30 messages up to Offset 1029, the total time spent has exceeded the max.poll.interval.ms limit (for example, because database queries slowed down).
  4. The Group Coordinator detects the delayed .poll() call, considers that consumer dead, and triggers a rebalance.
  5. The partition moves to another consumer. The new consumer reads the last valid committed offset on the broker, which is Offset 999 (because the previous batch commit would only be sent after all 50 messages finished processing).
  6. The new consumer reprocesses all 50 messages from Offset 1000, even though the first 30 messages were actually already successfully written to the database by the previous consumer.

The Idempotent Consumer Mechanism #

The Idempotent Consumer mechanism (often called the Deduplication Pattern) is an architecture pattern where the consumer application is designed to filter duplicate messages independently so that repeated executions of the same message don’t change the final system state beyond the first execution.

There are two main approaches we can apply to achieve this idempotency:

1. Database Unique Constraint / UPSERT (Natural Idempotency) #

If our business data entities naturally have a Natural Unique Key sent by the producer (for example, order_id, invoice_number, or payment_id), we can leverage the built-in safety features of RDBMS databases (like PostgreSQL, MySQL) or NoSQL (like MongoDB).

  • Unique Constraints: Mark the order_id column in the database as a UNIQUE PRIMARY KEY. If a consumer tries to write the same data a second time, the database cuts off the transaction and throws an Integrity Constraint Violation error exception.
  • UPSERT (Insert or Update): If we don’t want to throw an error but instead update the existing data, we can use the UPSERT command:
    • PostgreSQL: INSERT ... ON CONFLICT (order_id) DO UPDATE SET ... or DO NOTHING.
    • MySQL: INSERT INTO ... ON DUPLICATE KEY UPDATE ...
    • MongoDB: Using the db.collection.updateOne() operation with the upsert: true parameter.

2. Dedicated Deduplication Table #

In many real-world scenarios, our business logic isn’t as simple as storing data into one table. We may have to trigger third-party API calls, send notification emails to users, or run a series of complex queries that don’t have a single primary key.

To handle this scenario, we must create a Dedicated Deduplication Table (usually named processed_events or event_deduplication) in our transactional database.

  • How It Works:
    1. Every time a producer sends a message, it must include a unique UUID ID in the message header or payload (called the Event ID).
    2. When the consumer reads the message, it opens a new database transaction (BEGIN TRANSACTION).
    3. The consumer tries to insert that Event ID into the processed_events table.
    4. If the insert succeeds, the consumer continues executing the main business logic and updates the business database in the same transaction.
    5. The consumer commits the database transaction (COMMIT TRANSACTION).
    6. If the insert in step 3 fails because of a duplicate key, the consumer immediately rolls back the transaction (ROLLBACK), ignores that message, and continues to the next message.

Duplicate Message Detection Workflow #

The flow diagram below visualizes the decision flow our consumer application must go through when filtering messages using a deduplication table:

flowchart TD
    subgraph Alur_Pemrosesan
        Start["1. Consumer Calls poll()"] --> GetRecord["2. Get Message (Event ID: EV-888)"]
        GetRecord --> StartTx["3. Open Database Transaction (BEGIN)"]
        StartTx --> TryInsert["4. Try Inserting EV-888 into the processed_events table"]
        
        TryInsert --> Success{"Did the insert succeed?"}
        
        Success -- "Yes (Never Processed Before)" --> ProcessBiz["5. Execute Main Business Logic"]
        ProcessBiz --> CommitTx["6. Commit Database Transaction (COMMIT)"]
        CommitTx --> CommitOffset["7. Commit Offset to the Kafka Broker"]
        CommitOffset --> End["Done, Continue to the Next Message"]
        
        Success -- "No (Duplicate Detected)" --> RollbackTx["8. Roll Back the Database Transaction (ROLLBACK)"]
        RollbackTx --> SkipMessage["9. Ignore the Message / Log a Warning"]
        SkipMessage --> CommitOffset
    end

    style Success stroke:#fbc02d,stroke-width:2px
    style ProcessBiz stroke:#2e7d32,stroke-width:2px
    style SkipMessage stroke:#c62828,stroke-width:2px

Redis-Based Deduplication (In-Memory Distributed Cache) #

In very high-throughput applications (for example, processing hundreds of thousands of logistics logs per second), running INSERT queries to a relational database (RDBMS) for every message can trigger performance bottlenecks (disk I/O bottlenecks).

As an alternative, we can use distributed memory storage like Redis as a fast deduplication filter before data is sent to the RDBMS.

  • How It Works:
    • We leverage the atomic Redis command SET key value EX seconds NX.
    • The NX parameter ensures the key is only stored if it doesn’t already exist in Redis.
    • The EX parameter provides an automatic Time-To-Live (TTL) expiration limit so Redis memory doesn’t bloat unboundedly.
    • If Redis returns a success status, we continue processing to the database. If it fails (returns null or false), the message is immediately ignored because it’s proven to be a duplicate.
// EXAMPLE OF REDIS-BASED DEDUPLICATION IN JAVA (USING JEDIS)
String redisKey = "event:" + eventId;
// Store the Event ID with a 24-hour TTL (86400 seconds) only if the key doesn't exist (NX)
String result = jedis.set(redisKey, "processed", new SetParams().nx().ex(86400));

if ("OK".equals(result)) {
    // ✓ Safe: The Event ID has never been processed before
    executeDatabaseLogistics(record.value());
} else {
    // ✗ Duplicate detected in the Redis memory cache
    log.warn("Detected a duplicate message in the Redis cache: {}", eventId);
}
  • Trade-Off: This approach isn’t fully atomic-transactional (because Redis and the RDBMS are in separate systems). If the database write fails after the Redis key was successfully set, we must have a compensating transaction mechanism or delete the key in Redis so the message can be retried.

The Relationship with the Transactional Outbox Pattern #

To guarantee end-to-end data reliability, consumer-side duplication handling is often paired with the Transactional Outbox Pattern on the producer side.

  • The producer writes business data and event data to a local outbox table in one local database transaction.
  • A Debezium or CDC (Change Data Capture) component reads the outbox table and publishes it to the Kafka broker asynchronously.
  • Because CDC guarantees At-Least-Once delivery to the broker, consumers still must implement a deduplication table.
  • This synergy ensures that from the producer point to the consumer point, data is guaranteed not to be lost and the final system state is guaranteed consistent without transaction duplication.

Implementation Code: Idempotent Consumer Implementation #

Let’s directly compare the implementation code vulnerable to duplication with the robust solution code using RDBMS deduplication.

Java SDK Anti-Pattern: Data Storage Without Idempotency Protection #

The code below shows a common developer mistake assuming a regular database insert is safe enough because Kafka runs smoothly.

// ANTI-PATTERN: Writing data without message uniqueness filtering
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // Parse the JSON payload
            Order order = parseJson(record.value());
            
            // ✗ DON'T: Use a raw insert query without a unique constraint or checking.
            // If a rebalance or network failure happens before commitSync(),
            // this function creates a new order row in the database with the same ID,
            // triggering data duplication in the admin dashboard.
            saveOrderToDatabase(order.getId(), order.getCustomerId(), order.getAmount());
        }
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Java SDK Solution: Implementing a Transactional Deduplication Table #

The code below shows the recommended solution by wrapping the deduplication table insert process and the business write in one local database transaction using JDBC.

// CORRECT: Securing processing using transactional database transactions and constraints
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            
            // Get the unique Event ID from the Kafka header (or from the JSON payload)
            String eventId = getHeaderValue(record.headers(), "event_id");
            Order order = parseJson(record.value());

            if (eventId == null) {
                log.error("Message has no unique event_id, ignoring for safety.");
                continue;
            }

            try (Connection conn = dataSource.getConnection()) {
                // ✓ 1. Start the Local Database Transaction
                conn.setAutoCommit(false);

                try {
                    // ✓ 2. Try recording the Event ID into the processed_events deduplication table.
                    // The event_id column is the PRIMARY KEY.
                    try (PreparedStatement dedupStmt = conn.prepareStatement(
                        "INSERT INTO processed_events (event_id) VALUES (?)"
                    )) {
                        dedupStmt.setString(1, eventId);
                        dedupStmt.executeUpdate();
                    }

                    // ✓ 3. If successful, execute the main business logic in the same transaction
                    try (PreparedStatement bizStmt = conn.prepareStatement(
                        "INSERT INTO orders (id, customer_id, amount) VALUES (?, ?, ?)"
                    )) {
                        bizStmt.setString(1, order.getId());
                        bizStmt.setString(2, order.getCustomerId());
                        bizStmt.setDouble(3, order.getAmount());
                        bizStmt.executeUpdate();
                    }

                    // ✓ 4. Commit the entire database transaction as a whole
                    conn.commit();
                    log.info("Successfully processed event: {} and stored the business data.", eventId);

                } catch (SQLIntegrityConstraintViolationException e) {
                    // ✓ 5. DUPLICATE DETECTION: A primary key violation happened in step 2.
                    // Immediately roll back the database transaction so no duplicate data is written.
                    conn.rollback();
                    log.warn("Duplicate message detected for Event ID: {}. Safely ignoring the message.", eventId);
                } catch (Exception ex) {
                    conn.rollback();
                    log.error("A system error occurred, rolling back the transaction.", ex);
                    throw ex; // Throw the exception so the consumer loop stops & retry runs
                }
            } catch (SQLException sqle) {
                log.error("Database connection failure", sqle);
            }
        }
        
        // ✓ 6. Always commit offsets at the end of the batch
        if (!records.isEmpty()) {
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Summary #

  • At-Least-Once Side-Effect — Using the At-Least-Once guarantee in Apache Kafka ensures data is never lost, but carries the risk of message duplication from offset commit failures.
  • Idempotent Consumer Pattern — The mandatory architecture pattern on the consumer side ensuring repeated processing of the same message doesn’t damage the final business database state.
  • Natural Unique Key — The approach of using the business data’s built-in unique key column (like order_id) and relying on RDBMS Unique Constraints to cut off data duplication.
  • Deduplication Table — A special helper table (processed_events) for recording unique UUID Event IDs before executing complex business logic in one local database transaction.
  • Non-Transactional Side-Effects — Non-database external operations (like sending emails or deducting balances via third parties) must be secured by including an Idempotency Key when calling external APIs.
  • Atomic Transactionality — Deduplication safety is guaranteed by ensuring the event ID recording and business execution are wrapped in the same BEGIN ... COMMIT transaction block.

← Previous: At-Least-Once vs At-Most-Once Next: Poison Message Strategy →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact