	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Vector Databases > Kafka Consumer Offset Management

Kafka Consumer Offset Management

Author: Venkata Sudhakar

A Kafka offset is a number that tracks how far a consumer has read in a partition. When a consumer commits an offset, it tells Kafka "I have successfully processed everything up to this point." If the consumer crashes and restarts, it resumes from the last committed offset. Getting offset commits right determines whether your application processes every message exactly once, at least once, or risks losing messages.

Kafka provides two commit modes. Auto-commit (enable.auto.commit=true) commits offsets every 5 seconds in the background - simple, but it can commit an offset before your processing is done, causing message loss on restart. Manual commit gives you full control: you call commitSync() or commitAsync() only after you have successfully processed the message, guaranteeing at-least-once delivery.

The below example shows auto-commit vs manual commit and why the timing of the commit matters.

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class OffsetExample {

// Pattern 1: Auto-commit - simple but risky
    // Kafka commits every 5 seconds regardless of processing status
    static void autoCommit() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-processor");
        props.put("enable.auto.commit", "true");       // commits every 5s
        props.put("auto.commit.interval.ms", "5000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> r : records) {
                    processOrder(r.value()); // if app crashes here, offset may already be committed
                }
                // Risk: auto-commit fires at 5s mark even if processOrder() hasn't finished
            }
        }
    }

// Pattern 2: Manual commit - commit AFTER processing is done
    static void manualCommit() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-processor");
        props.put("enable.auto.commit", "false");      // disable auto-commit
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> r : records) {
                    processOrder(r.value()); // process first
                }
                consumer.commitSync(); // THEN commit - safe to advance the offset
            }
        }
    }

static void processOrder(String order) {
        System.out.println("Processing: " + order);
    }
}

It gives the following output,

Processing: {"orderId":"ORD-1001","amount":99.99}
Processing: {"orderId":"ORD-1002","amount":49.99}

# With manual commit:
# - Offset committed AFTER both messages processed
# - If app crashes mid-batch, both messages are re-processed on restart
# - At-least-once delivery guaranteed

# With auto-commit:
# - Offset committed at 5s mark regardless of processing state
# - If app crashes after commit but before processing, messages are lost

Use manual commitSync() for critical workloads where losing a message is not acceptable. commitAsync() is faster but does not retry on failure - use it inside a batch loop and only use commitSync() on shutdown or after a complete batch. The tradeoff is always between safety (manual sync) and throughput (auto or async).

Send your comments, suggestions or queries regarding this site to [email protected].