At-Least-Once #

In the world of distributed system architecture, data loss is the worst scenario that architects and application developers must avoid by all means. To guarantee that every message produced and sent by the system always gets successfully processed at the destination system, Apache Kafka provides the At-Least-Once delivery guarantee model. This is the default and most popular guarantee level used in the Kafka ecosystem. Through At-Least-Once semantics, Kafka gives an absolute commitment to data reliability (zero data loss). Every message is guaranteed to be successfully read and processed by the downstream system at least once. The logical consequence of this guarantee is the potential appearance of duplicate messages or data duplication on the consumer side if a network failure or application crash happens right before the read confirmation (offset commit) is stored on the broker.

The Core Philosophy of At-Least-Once and the Zero Data Loss Concept #

The core philosophy behind At-Least-Once semantics is that absolute reliability is more valuable than the cost of handling duplicate data. In traditional monolithic systems, we can use local database transactions (ACID) to ensure data write operations and status updates run atomically. However, in microservices architecture and large-scale distributed systems, ACID transactions across network boundaries are very slow and prone to partial failures (split-brain).

Kafka solves this problem by implementing a two-way guarantee strategy:

  1. Producer Side: The producer doesn’t consider a message delivery successful before receiving written proof (ACK confirmation) that the message has been safely stored on several replica brokers. If the ACK fails to arrive due to network disruptions, the producer keeps trying to resend that message.
  2. Consumer Side: Consumers are strictly forbidden from informing the broker that they’ve finished reading a message before the application business logic (like writing to a relational database, calling third-party APIs, or updating caches) is confirmed to have run successfully without error exceptions. This pattern is known as Process-First, Commit-Second.
sequenceDiagram
    autonumber
    participant Prod as Producer Client
    participant Broker as Kafka Broker (ISR)
    participant Cons as Consumer Client
    participant DB as Client Database

    Note over Prod, Broker: "Producer Side: Storage Guarantee (acks=all)"
    Prod->>Broker: ProduceRequest (Message X)
    Note over Broker: Broker writes to Leader & Followers (ISR)
    Broker-->>Prod: Success ACK (Message X safe on disk)

    Note over Broker, DB: "Consumer Side: Process-First (At-Least-Once)"
    Cons->>Broker: poll() fetches Message X (Offset 500)
    Cons->>DB: Write Message X to the Database
    DB-->>Cons: Write Success
    Note over Cons: CRASH! (Server dies suddenly before sending the offset commit)

    Note over Cons, Broker: "Process After Recovery"
    Note over Cons: New consumer / restart re-reads
    Cons->>Broker: poll() from Offset 500 (Broker doesn't know about the earlier crash)
    Cons->>DB: Write Message X to the Database (DUPLICATE!)
    DB-->>Cons: Write Success
    Cons->>Broker: commitSync(Offset 500)
    Broker-->>Cons: Commit ACK (Offset 500 safely stored)

Through the sequence diagram above, we can clearly see that in step 6, the data was successfully written to the client’s external database. However, because a sudden crash happened in step 7 before the offset commit was sent, after the consumer recovers in step 8, it pulls the same message (Offset 500) again from the broker. This causes rewriting the same data to the database in step 10, producing data duplication.

Producer Configuration for the Zero-Data-Loss Guarantee #

To ensure the producer never loses messages before they’re stored in the Kafka cluster, we must configure the producer durability parameters as follows:

1. acks=all (or acks=-1) #

This is the most critical parameter for guaranteeing durability.

  • How It Works: The producer only considers a message delivery successful after the partition leader broker successfully replicates that message to all followers in the In-Sync Replicas (ISR) list.
  • Synergy with min.insync.replicas: This configuration must be paired with the broker parameter min.insync.replicas (for example, set to 2 on a topic with Replication Factor = 3). If the number of active synchronized replicas is below this minimum limit, the leader broker refuses to write new data and triggers the NotEnoughReplicasException error on the producer side.

2. retries=2147483647 (Integer.MAX_VALUE) #

Since Kafka 2.0, the default retries value is the maximum integer value.

  • How It Works: If temporary network disruptions happen (for example, socket connection pauses, GC pauses on the leader broker, or new leader election transition processes), the producer actively retries sending that message until it succeeds.
  • Delivery Timeout: The maximum limit of these retry attempts is controlled by the delivery.timeout.ms property (by default set to 120000 ms or 2 minutes). As long as this timeout isn’t exceeded, the producer doesn’t give up resending the message.

3. max.in.flight.requests.per.connection #

By default, this parameter is 5. If a retry happens on one failed batch request while the next batch on the network was already successfully sent, the message order on the broker can become random (out-of-order).

  • Solution Without Idempotence: Set this property to 1 (significantly reducing throughput).
  • Modern Solution: Use enable.idempotence=true (highly recommended because it keeps message order consistent even with max.in.flight up to 5).

Consumer Configuration: The Process-First Pattern #

To guarantee At-Least-Once semantics on the consumer side, the offset commit responsibility is entirely on our application code logic. We must reject imprecise time-based automatic commits.

The consumer property that must be set:

  • enable.auto.commit=false: Requires us to trigger offset commit delivery manually via code.

The Correct Consumer Workflow (Process-First) #

  1. The client calls .poll(Duration) to fetch a list of data records from the broker.
  2. Loops through to process each record sequentially.
  3. Stores transaction status or record processing results to the database or calls external HTTP APIs.
  4. Only after all processing succeeds without exceptions, call the consumer.commitSync() or consumer.commitAsync() function to inform the broker that the data batch has been fully processed.
flowchart TD
    Poll["Poll Data"] --> Process["Process Business Logic & Save to DB"] --> Commit["Commit Offset to Broker"]
    Process -- "Application Crash" --> ReRead["Data Re-read from the Broker"]

If an error happens while processing the 3rd record out of 10 fetched records, our code must throw an exception and stop the process before calling the commit function. This way, the broker still records the old offset. When a new consumer takes over that partition, it reprocesses the 10 records from the start, ensuring the failed 3rd record isn’t missed.

Detailed Chronology of Data Duplication in Production #

Let’s trace real scenarios of how duplicate messages can enter our database when using At-Least-Once semantics.

Scenario 1: Producer ACK Lost on the Network #

  1. The producer sends the ProduceRequest(payload="Message A") message to the Leader Broker.
  2. The broker writes the message to the local disk log, copies it to ISR followers, and creates a success ACK.
  3. Right before the ACK is sent to the producer’s network card, the network switch on the broker rack experiences a momentary power outage. The ACK data packet fails to reach the producer client.
  4. The producer client waits until the request.timeout.ms limit (default 30 seconds) passes.
  5. Because it didn’t receive the ACK, the producer considers the message failed to send due to network disruption.
  6. The producer does an automatic retry and resends ProduceRequest(payload="Message A").
  7. The broker receives that message. Because the broker considers this a legitimate new request (if idempotence is disabled), the broker writes “Message A” to the partition disk log at the next offset (for example, offset 101).
  8. The broker sends a new success ACK. The producer receives it fine.
  9. Result: Inside the Kafka topic there are now two identical messages (“Message A”) at offset 100 and offset 101. Both are read by consumers, producing duplicate processing.

Scenario 2: Consumer Crash during Rebalance #

  1. Consumer C1 calls .poll() and receives 100 records from partition 2, starting from offset 1000 to 1099.
  2. C1 successfully processes 99 records (offsets 1000 to 1098) and saves them to the MySQL database.
  3. Right before processing the last record (offset 1099), the broker detects that a new consumer C2 has joined the group. The Group Coordinator triggers a Rebalance process.
  4. The old rebalance protocol (Eager Rebalance) forces C1 to release partition 2 ownership before it gets a chance to call commitSync().
  5. Partition 2 is reallocated to the new consumer C2.
  6. C2 starts reading data from partition 2. Because the last committed offset on the broker is still 1000, C2 re-fetches records from offset 1000 to 1099.
  7. Result: The 99 records already successfully processed by C1 are rewritten to the MySQL database by C2, triggering 99 rows of data duplication if there’s no idempotent filtering on the database side.

Rescue Tactics: How to Deal with Message Duplication? #

Because data duplication is an unavoidable logical consequence of At-Least-Once semantics, our consumer applications must have defenses to filter those duplicate messages. Here are three commonly used rescue tactics:

1. Idempotent Business Logic Design #

An idempotent operation is an operation that, if run repeatedly with the same input, produces the same final result without changing system state more than once.

  • Non-Idempotent Example: UPDATE users SET balance = balance + 10000 WHERE id = 1. If run twice because of message duplication, the user’s balance increases by 20,000 (wrong!).
  • Idempotent Example: UPDATE users SET balance = 50000 WHERE id = 1. No matter how many times it runs, the final balance stays 50,000.

2. Unique Constraint Validation #

Leveraging the unique key/primary key constraint feature on relational (SQL) databases or document (NoSQL) databases.

  • How It Works: Before writing transaction data, create a combined unique column (for example, combining transaction_id and event_timestamp) as the Primary Key.
  • Handling: If a duplicate message arrives, the database throws a unique key violation exception (UniqueConstraintViolationException). Our consumer code just catches this exception, records a warning log, and safely ignores it (continuing to the offset commit step).

3. Redis Deduplication Table (Idempotent Consumer Pattern) #

If our destination database doesn’t naturally support unique constraints (like writing to Elasticsearch or calling third-party SMS APIs), we can use Redis as a fast deduplication filter.

  • How It Works: Every time a message is received, the consumer checks whether message_id or event_uuid already exists in Redis using the atomic SETNX (Set if Not Exists) operation with a Time to Live (TTL) expiration limit.
  • Logic: If SETNX returns true, process the message and commit. If it returns false, silently discard that message because it was already processed before by another thread.

Java SDK Implementation Code: The Best Manual Commit #

Here’s an example of a Kafka consumer implementation using the Java SDK that applies At-Least-Once semantics through manual commit handling, network error handling, and clean connection shutdown (graceful shutdown).

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class AdvancedAtLeastOnceConsumer {
    private static final Logger log = LoggerFactory.getLogger(AdvancedAtLeastOnceConsumer.class);
    private static final AtomicBoolean running = new AtomicBoolean(true);
    private static KafkaConsumer<String, String> consumer;

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // =====================================================================
        // ZERO DATA LOSS CONFIGURATION (AT-LEAST-ONCE)
        // =====================================================================
        // ✓ Must disable auto-commit so we can ensure processing succeeds first
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        
        // ✓ If no offset exists on the broker, start reading from the oldest position
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("customer-orders"));

        // Set up a Shutdown Hook for clean connection shutdown
        final Thread mainThread = Thread.currentThread();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("Detecting termination signal, stopping the poll loop...");
            running.set(false);
            // Trigger a WakeupException on the main thread stuck in .poll()
            consumer.wakeup();
            try {
                mainThread.join();
            } catch (InterruptedException e) {
                log.error("Failed to wait for the main thread to finish", e);
            }
        }));

        try {
            while (running.get()) {
                // Fetch a message batch from the broker
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                if (!records.isEmpty()) {
                    log.info("Received {} records from poll.", records.count());
                    boolean batchSuccess = true;

                    for (ConsumerRecord<String, String> record : records) {
                        try {
                            // ✓ PROCESS FIRST: Run the important business logic (Save to DB / calculate transactions)
                            processOrderRecord(record);
                        } catch (Exception e) {
                            // ✗ DON'T: Continue processing the next record if a fatal database error happens.
                            // We must mark the batch as failed so the offset isn't committed to the broker.
                            log.error("✗ Fatal error while processing offset {}. Stopping the batch.", record.offset(), e);
                            batchSuccess = false;
                            break;
                        }
                    }

                    // ✓ COMMIT SECOND: Only send the offset commit if ALL records were successfully processed
                    if (batchSuccess) {
                        try {
                            // Using commitSync() synchronously for strong block processing guarantees.
                            // commitSync() blocks the thread until the broker replies successfully.
                            consumer.commitSync();
                            log.info("✓ Successfully committed the current batch offset.");
                        } catch (CommitFailedException e) {
                            log.error("✗ Failed to commit the offset to the broker (Rebalance happening or timeout):", e);
                            // Failure here means data will be reprocessed on the next poll (safe from data loss).
                        }
                    } else {
                        // Rescue Scenario: If batch processing fails, we must rewind the read pointer
                        // to the lowest offset of the failed batch so no offset gaps happen.
                        seekToFirstUncommittedOffset(records);
                    }
                }
            }
        } catch (WakeupException e) {
            // Ignore this exception because it's deliberately triggered during application shutdown
            log.info("Poll loop cleanly stopped via the Wakeup API.");
        } catch (Exception e) {
            log.error("An unexpected exception occurred in the consumer loop", e);
        } finally {
            try {
                // Always do a final synchronous commit before closing the connection
                consumer.commitSync();
            } catch (Exception e) {
                log.warn("Failed to do a final commit during consumer shutdown.");
            } finally {
                consumer.close();
                log.info("Consumer connection cleanly closed.");
            }
        }
    }

    private static void processOrderRecord(ConsumerRecord<String, String> record) throws Exception {
        log.info("Processing order: Key={}, Offset={}, Value={}", record.key(), record.offset(), record.value());
        // Simulate a random database error for data resilience testing
        if (record.value().contains("INVALID_PAYMENT")) {
            throw new Exception("Database connection dropped or invalid payment format.");
        }
        // Successful business logic
    }

    private static void seekToFirstUncommittedOffset(ConsumerRecords<String, String> records) {
        // Logic to rewind the consumer read pointer to the lowest offset of the failed batch
        records.partitions().forEach(partition -> {
            long minOffset = records.records(partition).get(0).offset();
            log.warn("Rewinding partition {} read pointer to offset {}", partition, minOffset);
            consumer.seek(partition, minOffset);
        });
    }
}

Data Delivery Guarantee Comparison #

CharacteristicAt-Most-OnceAt-Least-OnceExactly-Once
Data Loss RiskVery HighZero LossZero Loss
Duplication RiskZero (No Duplicates)High (Potential Duplicates)Zero (No Duplicates)
Throughput & LatencyMost MaximumMedium-HighLow-Medium (Transaction Overhead)
Client Operation OrderCommit-FirstProcess-FirstAtomic Transactional
Deduplication NeedsNot NeededVery MandatoryNot Needed (Kafka-to-Kafka)

Summary #

  • At-Least-Once — The delivery guarantee where every message is confirmed successfully processed at least once by the consumer, prioritizing data safety above duplication risk.
  • Process-First, Commit-Second — The code writing strategy where consumers process all business logic and database storage first before sending offset commit signals to the broker.
  • Producer-Side Durability — Achieved by configuring acks=all and maximum retries, ensuring messages are safely copied to the In-Sync Replicas (ISR) member list before being considered successful.
  • Data Duplication Triggers — Caused by ACK packets lost on the network during producer delivery, or consumer application crashes midway after database writes but before offset commits.
  • Database Deduplication Tactics — Using unique/primary key constraints on SQL databases to automatically catch unique key violation exceptions when duplicate messages arrive.
  • Redis In-Memory Deduplication — Implementing fast cache filters using the atomic SETNX command to filter message IDs already executed before by other threads.
  • Graceful Shutdown Hook — Applying the consumer.wakeup() API to interrupt consumer threads blocking on the network, preventing wild rebalances when the application shuts down.

← Previous: At-Most-Once Next: Exactly-Once →

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