Auto vs Manual Commit #

When our consumer application reads messages from Apache Kafka, the broker doesn’t track which messages have been successfully processed by the consumer. The broker only records the last read position marker of our consumer group for each partition. This position marker is stored in a special internal topic called __consumer_offsets and is known as the Committed Offset. Every time our consumer restarts or a partition reassignment process (rebalance) happens, the new consumer reads its starting processing position from this last committed offset. Therefore, the timing and safety of sending offset data to the broker is the most critical factor determining our data transmission reliability. We face a fundamental choice: using the convenient but dangerous automatic commit mechanism (Auto Commit), or controlling offset storage manually (Manual Commit) to guarantee our business data integrity.

The Auto-Commit Mechanism #

By default, the Kafka consumer client is configured to use the automatic commit mechanism with the enable.auto.commit=true property.

How Auto-Commit Works #

When this feature is active, the consumer periodically announces the highest offset obtained from .poll() calls to the coordinator broker. This periodic time interval is controlled by the auto.commit.interval.ms property with a default value of 5,000 ms (5 seconds).

This mechanism runs passively inside the main poll loop thread:

The application calls consumer.poll(). The consumer client internally checks whether the time since the last commit has exceeded the 5-second limit.
If the 5-second limit has passed, the consumer attaches a last-offset commit request to the next network I/O request (for example, when sending a heartbeat or making a new fetch request).
The broker updates the offset in __consumer_offsets, and the 5-second cycle restarts from zero.

This mechanism is very popular because of its simplicity — developer teams don’t need to write a single line of code to manage offsets, because the Kafka client handles everything in the background. However, this convenience hides fatal failure risks that can damage our data quality in production.

The Deadly Danger of Using Auto-Commit in Production #

Although it simplifies code writing, leaving the enable.auto.commit=true property active in production environments with high-value data traffic is highly not recommended. This passive time-based mechanism can trigger two data disaster scenarios: Data Loss and Data Duplication.

Scenario 1: Data Loss #

Imagine the following scenario happens in our consumer application:

The consumer calls .poll() and receives 100 messages (Offsets 100 to 199).
These messages are temporarily stored in the application’s heap memory queue.
The main thread starts processing these messages one by one. The business logic for each message takes 100 milliseconds (10 seconds total to finish the entire batch).
Critical Event: Exactly at the 5th second of processing (when the application has successfully processed 50 messages up to Offset 149), the main thread calls the next .poll() to keep the loop alive. Because the 5-second window has expired, the consumer client automatically sends an offset commit to the broker for Offset 199 (because Offset 199 is the highest offset fetched in step 1). The broker records Offset 199 as the successful commit position.
At the 6th second, our application suddenly crashes (e.g., from OutOfMemory, power outage, or restart by the Kubernetes scheduler). Processing of messages Offset 150 to 199 fails completely and never executes.
When the application container recovers and the consumer restarts, the consumer requests the starting offset from the broker. The broker returns Offset 199.
Final Result: The consumer starts reading from Offset 200. Messages from Offset 150 to 199 are permanently lost from our application processing cycle without any recorded error trace.

Scenario 2: Data Duplication #

The opposite of the scenario above: if our application successfully processes 90 messages (up to Offset 189) in 4 seconds, then crashes before the 5-second interval expires:

The broker hasn’t received the automatic offset update yet, so the last committed offset on the broker is still lagging at Offset 99.
When the new consumer starts, it’s forced to re-read data from Offset 100.
Final Result: Messages Offset 100 to 189 are processed a second time, triggering data duplication in our downstream database.

The Best Alternative: Manual Commit #

To guarantee the maximum data safety level, we must disable the automatic commit feature and fully control when offsets should be committed manually:

enable.auto.commit=false

By disabling auto-commit, we can apply strict business rules: Offset commits may only be sent to the broker if and only if the entire set of messages in that poll batch has been fully processed and successfully saved to our main database storage.

To trigger commits manually, the Java client SDK provides two main methods: commitSync() and commitAsync().

Dissecting commitSync() vs commitAsync() #

Choosing between these two manual methods is an architectural decision involving a trade-off between throughput speed and error recovery reliability.

1. `commitSync()` (Synchronous Manual Commit) #

This method acts blocking. When called, the consumer main thread stops spinning and waits until the coordinator broker finishes writing the offset to the __consumer_offsets topic and returns a success confirmation response packet.

Advantages: High safety. If a transient failure occurs (like a busy network or new leader election), commitSync() automatically retries until the timeout limit is reached before finally throwing an exception to the application.
Disadvantages: High transmission latency. Waiting for a network ACK on every loop iteration limits our consumer’s maximum throughput.

2. `commitAsync()` (Asynchronous Manual Commit) #

This method acts non-blocking. The consumer thread only sends the commit request to the network, then immediately continues the loop to process the next data without ever waiting for the broker’s response.

Advantages: Very high throughput and minimal latency.
Disadvantages: No automatic retry on failure. Why? Because asynchronous retry can trigger the Commit Race Condition problem.

The Commit Race Condition Problem #

Imagine our consumer sends two sequential asynchronous commit requests:

The consumer calls commitAsync(Offset 100). This request gets temporarily delayed on the network.
The consumer processes the next batch and calls commitAsync(Offset 200). This request reaches the broker faster. The broker records the last offset as 200.
The first commitAsync(Offset 100) request finally gets through the network congestion and reaches the broker.
Final Result: If commitAsync() did automatic retry for old failures, it would incorrectly overwrite offset 200 back to 100. Therefore, commitAsync() is designed to never retry on failure.

The Best Combination Pattern (The Standard Commit Pattern) #

To get the speed advantage of commitAsync() while retaining the reliability of commitSync(), the industry standard recommends using the following combined pattern:

Use commitAsync() inside the main loop for maximum throughput performance without blocking the processing thread.
Wrap the entire loop block in a try-catch-finally handler.
In the finally block (triggered when the application receives a shutdown signal), call commitSync() once forcibly to ensure the last remaining offset before the application dies is truly safely stored on the broker.

Mermaid Diagram: Auto Commit vs Manual Commit Flow #

The following sequence diagram compares how offset state is managed between the data-loss-prone Auto Commit model versus the safe Manual Commit model:

sequenceDiagram
    autonumber
    participant App as Application Thread
    participant Cons as Kafka Consumer Client
    participant Broker as Kafka Broker Coordinator
    
    Note over App, Broker: "Scenario 1: The Danger of Auto-Commit (enable.auto.commit=true)"
    App->>Cons: poll(100ms)
    Cons->>Broker: Fetch Data (Offset 100-110)
    Broker-->>Cons: Return 10 records
    Cons-->>App: Return records
    App->>App: "Start Business Processing (Takes 6 seconds)"
    Note over Cons, Broker: "5th second: Auto-Commit triggered automatically on the next poll"
    App->>Cons: poll(100ms)
    Cons->>Broker: CommitOffsetRequest (Offset 110)
    Broker-->>Cons: Commit ACK
    Note over App: "6th second: Application Crashes / OOM while processing data!"
    Note over App: "On restart, the new consumer reads from Offset 110 (Data 100-109 LOST!)"
    
    Note over App, Broker: "Scenario 2: Manual Commit Control (enable.auto.commit=false)"
    App->>Cons: poll(100ms)
    Cons->>Broker: Fetch Data (Offset 200-210)
    Broker-->>Cons: Return 10 records
    Cons-->>App: Return records
    App->>App: "Business Processing Success (Write to DB)"
    App->>Cons: commitSync()
    Cons->>Broker: CommitOffsetRequest (Offset 210)
    Broker-->>Cons: Commit ACK (Offset safely stored)

Java Implementation: Setting the Best Manual Commit Pattern #

Here’s a Java code comparison between the data-loss-prone auto-commit usage (anti-pattern) and the recommended manual commit combination implementation (commitAsync + commitSync) for production stability:

// ANTI-PATTERN: Relying on time-based auto-commit for financial transaction data
public class UnsafePaymentConsumer {
    public void startConsuming(Properties props) {
        // ✗ Very dangerous: Using default auto-commit for important transactions
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("payments"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // If this process crashes midway, data after the 5th second gets auto-committed
                    // and triggers permanent data loss
                    executePayment(record.value());
                }
            }
        }
    }
    private void executePayment(String val) {}
}

// CORRECT: Using the safe combination of manual commitAsync() and commitSync()
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SafeTransactionalConsumer {
    public void consumeSecurely(Properties props) {
        // ✓ CORRECT: Disable auto commit to take full control
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

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

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    processPayment(record.value());
                }
                
                // ✓ CORRECT: Use commitAsync() to keep throughput high in the main loop
                // We pass an empty callback or simple logger to monitor network errors
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(java.util.Map<org.apache.kafka.common.TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            // Log if an asynchronous commit failure occurs
                            System.err.println("Failed to commit offsets asynchronously: " + exception.getMessage());
                        }
                    }
                });
            }
        } catch (WakeupException e) {
            System.out.println("Consumer woken up for shutdown...");
        } catch (Exception e) {
            System.err.println("Fatal error in the consumption process: " + e.getMessage());
        } finally {
            try {
                // ✓ CORRECT: Use commitSync() blocking in the finally block
                // This ensures the last offset before the application dies is truly written to the broker
                System.out.println("Performing the final synchronous commit before exiting...");
                consumer.commitSync();
            } finally {
                consumer.close();
                System.out.println("Consumer successfully closed gracefully.");
            }
        }
    }

    private void processPayment(String payload) {
        System.out.println("Successfully processed payment: " + payload);
    }
}

Understanding CommitFailedException during Rebalance #

One of the biggest challenges when using manual commits in production is handling CommitFailedException. This exception is thrown synchronously by commitSync() (or reported through the asynchronous callback on commitAsync()) when the coordinator broker detects that the partition our consumer is trying to commit has been allocated to another consumer.

Why Does This Happen? #

This scenario is most often triggered by exceeding the max.poll.interval.ms processing time limit.

The consumer calls .poll(), gets a data batch, then the main thread processes the data slowly (exceeding the default 5-minute limit).
The Coordinator broker concludes that consumer is dead (stuck/livelock). The Coordinator removes the consumer from the group and triggers a Rebalance process.
That consumer’s partitions are allocated to new healthy consumers. The new consumer starts reading data from the last offset recorded on the broker.
The old consumer finally finishes processing its slow data and calls consumer.commitSync().
The Coordinator broker rejects the request and throws CommitFailedException because the partition ownership status has changed hands.

The Correct Handling Method #

We must not try to retry a commit that failed from CommitFailedException. We must catch this exception, record an audit log, discard the local processing state, and let the new consumer take over. Forcibly writing offsets from a removed consumer only damages the offset data belonging to the new consumer.

Granular Commits: Specifying Specific Offsets Per Partition #

By default, calling commitSync() or commitAsync() without arguments commits all the last offsets from all partitions returned by the previous .poll() call. This approach is sometimes considered insufficiently precise for very dense throughput systems.

To provide full control, the Java client provides commit functions accepting specific parameters:

// Committing offsets granularly for specific partitions only
consumer.commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

Granular Commit Advantages #

By explicitly mapping offsets per partition, we can do staged commits (mid-batch commits). For example, if one .poll() call returns records from 3 partitions (Partition 0, Partition 1, and Partition 2), our application can:

Process all records specifically for Partition 0.
Immediately commit offsets only for Partition 0.
Continue processing for Partition 1 and so on.

This drastically minimizes the number of messages that must be reprocessed (duplication) if our application crashes in the middle of processing a large batch.

Here’s a granular commit implementation example in Java:

// ✓ CORRECT: Committing offsets granularly per partition to minimize data duplication
public void consumeGranularly(KafkaConsumer<String, String> consumer) {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            
            for (ConsumerRecord<String, String> record : partitionRecords) {
                processRecord(record);
            }
            
            // Get the offset of the last record in this partition, plus 1 (the next read position)
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            
            // Create a commit map for this specific partition
            Map<TopicPartition, OffsetAndMetadata> commitMap = Collections.singletonMap(
                partition, 
                new OffsetAndMetadata(lastOffset + 1)
            );
            
            // ✓ Commit offsets only for the partition that has been fully processed
            consumer.commitSync(commitMap);
        }
    }
}

Summary #

Offset Commit: The mechanism for recording the consumer group’s last read position on the Kafka broker, stored in the internal __consumer_offsets topic.
Auto-Commit Danger: The time-based auto-commit feature (5s) is prone to triggering data loss disasters if the application crashes mid-batch processing.
Manual Control: Disabling auto-commit (enable.auto.commit=false) is the mandatory production standard to guarantee precise data processing.
commitSync: A blocking commit operation guaranteeing offset write safety with automatic retry if temporary network disruptions occur.
commitAsync: A non-blocking commit operation maintaining maximum throughput speed by eliminating network response wait time.
Combined Pattern: The best design pattern uses commitAsync() in the main loop and seals it with commitSync() at shutdown.

← Previous: Poll Loop Next: Offset Management →