When working with Apache Kafka, managing how consumer offsets are committed is essential for ensuring that messages are processed once and only once, especially in the context of reprocessing messages after a failure. Manually committing offsets provides greater control over this process, as opposed to the default auto-commit configuration.

Understanding Offset Committing

In Kafka, an offset is a pointer to the last message that a consumer has read and processed from a partition of a topic. Committing the offset means that the consumer records the offset in Kafka such that if the consumer fails, it can restart reading from the last committed offset.

Manual vs. Automatic Offset Committing

The key difference between manual and automatic offset committing lies in who controls the commit process and the potential for finer-grained control:

• Automatic Committing: Kafka periodically commits the offsets according to the configuration auto.commit.interval.ms. This is simple but can lead to issues like duplicate processing if a consumer crashes between a message process and offset commit.
• Manual Committing: The application explicitly commits the offsets. This usually happens after the message has been processed, providing much tighter control over when the commit happens, thus reducing the chances of message duplication.

Manual Offset Committing Methods

Kafka provides various ways to commit offsets manually:

1. Synchronous Commit: commitSync() is a blocking call that commits the highest offset of messages that have been processed by the consumer so far. This method guarantees that the offset is committed but can reduce throughput due to its blocking nature.
2. Asynchronous Commit: commitAsync() commits offsets without blocking processing. Errors and successful commits are handled via callbacks. This method provides higher throughput but handling errors can be complex.

Implementing Manual Offset Commit with Examples

Below is a basic example in Java demonstrating how to manually commit offsets using both synchronous and asynchronous methods.

1import org.apache.kafka.clients.consumer.ConsumerRecord;
2import org.apache.kafka.clients.consumer.ConsumerRecords;
3import org.apache.kafka.clients.consumer.KafkaConsumer;
4
5import java.util.Arrays;
6import java.util.Properties;
7
8public class Consumer {
9    public static void main(String[] args) {
10        Properties props = new Properties();
11        props.put("bootstrap.servers", "localhost:9092");
12        props.put("group.id", "test");
13        props.put("enable.auto.commit", "false"); // Disable auto-commit
14        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
15        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
16
17        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
18            consumer.subscribe(Arrays.asList("my-topic"));
19
20            while (true) {
21                ConsumerRecords<String, String> records = consumer.poll(100);
22                for (ConsumerRecord<String, String> record : records) {
23                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
24                    // Process record
25
26                    // Synchronous commit
27                    consumer.commitSync();
28
29                    // Asynchronous commit
30                    consumer.commitAsync((offsets, exception) -> {
31                        if (exception != null) {
32                            System.err.println("Commit failed for offsets " + offsets);
33                            exception.printStackTrace();
34                        }
35                    });
36                }
37            }
38        }
39    }
40}

Best Practices for Manual Offset Committing

• Use asynchronous commits during normal processing: This maximizes throughput and minimizes consumer latency.
• Use synchronous commits for shutdown: This ensures that all processed messages are committed correctly before shutting down.
• Handle exceptions wisely: Carefully manage exceptions during commits, especially with asynchronous commits.

Summary Table

Conclusion

Manually committing offsets in Kafka offers precise control over message delivery semantics, especially in distributed systems where reliability and fault tolerance are critical. By understanding and implementing either of the manual committing methods appropriately, developers can ensure that their Kafka consumer applications are robust and capable of handling failures gracefully.