How do you commit offsets manually in a Kafka consumer implementation using Java?

Committing offsets manually in a Kafka consumer implementation using Java can be useful when you want more fine-grained control over the offset commit process. Here’s an example of how to commit offsets manually in a Kafka consumer implementation using Java:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my_group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my_topic"));

while (true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord record : records) {
        // Process the message here
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map offsets, Exception exception) {
            if (exception != null) {
                // Handle the exception here
            }
        }
    });
}

consumer.close();

In this example, we have defined the properties for the Kafka consumer and subscribed to the “my_topic” topic using the subscribe method.

When messages are received, they are processed in the same way as in the previous examples. Finally, we have used the commitAsync method to commit the offsets of the processed messages. The commitAsync method takes an OffsetCommitCallback object as a parameter, which is called when the offset commit is complete. The OffsetCommitCallback includes the offsets that were committed and any exceptions that occurred during the commit process.

By using the commitAsync method, you can commit offsets manually in a Kafka consumer implementation using Java and gain more fine-grained control over the offset commit process. However, it’s important to handle exceptions that may occur during the commit process, such as CommitFailedException or RetriableCommitFailedException, and implement appropriate error handling and retry strategies as needed.