Kafka Consumer Using Java

Implementation of Kafka Consumer using Java

package com.abc;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerTest {

 public static void main(String[] args) throws InterruptedException, ExecutionException{
  //Create consumer property
  String bootstrapServer = "localhost:9092";
  String groupId = "my-first-consumer-group";
  String topicName = "my-first-topic";
  
  Properties properties = new Properties();
  properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
  properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
  properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
  
  //Create consumer
  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
  
  //Subscribe consumer to topic(s)
  consumer.subscribe(Collections.singleton(topicName));
  
  
  //Poll for new data
  while(true){
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
   
   for(ConsumerRecord<String, String> record: records){
    System.out.println(record.key() + record.value());
    System.out.println(record.topic() + record.partition() + record.offset());
   }
   
   //Commit consumer offset manually (recommended)
   consumer.commitAsync();
  }
  
 }
}

Consumer Groups and group.id

Every consumer belongs to a consumer group, identified by ConsumerConfig.GROUP_ID_CONFIG ("group.id"). Kafka guarantees that within a group, each partition of the subscribed topic is consumed by exactly one consumer at a time — so if my-first-topic has 6 partitions and you start 3 consumer instances with the same group.id, each one is assigned roughly 2 partitions. Start a 4th instance and the group rebalances, redistributing partitions across all 4. Consumers with different group.id values are independent of each other and each gets its own full copy of every partition’s data.

This is the mechanism that gives you both horizontal scaling (add more consumer instances, up to one per partition, to process faster) and pub/sub-style fan-out (give each application its own group id).

auto.offset.reset Only Applies When There’s No Committed Offset

ConsumerConfig.AUTO_OFFSET_RESET_CONFIG ("earliest" here) decides where to start reading only when the broker has no valid committed offset for that consumer group and partition — either because the group has never consumed from this topic before, or because the previously committed offset has aged out (governed by the broker’s offsets.retention.minutes) or fallen outside the log’s retained range. If a committed offset already exists, the consumer resumes from it regardless of what auto.offset.reset says. A common mix-up: restarting a consumer with a new group.id to “start fresh” will trigger auto.offset.reset again, while restarting with the same group.id will not, even after code changes.

Manual Commits and Delivery Semantics

This example sets ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false and calls consumer.commitAsync() after processing each batch of records — that ordering (process, then commit) gives you at-least-once delivery: if the process crashes after handling a record but before the commit lands, the next consumer to take that partition re-reads and reprocesses it. Committing before processing would give you at-most-once instead (a crash mid-processing means the record is skipped for good). Kafka’s consumer API doesn’t give you exactly-once processing on its own — that requires either idempotent downstream writes or the transactional producer/consumer APIs.

commitAsync() doesn’t retry on failure (retrying could commit a stale offset out of order if a later commit already succeeded), so production code typically passes it a callback to at least log failures, and calls the blocking commitSync() once during shutdown to make sure the final offset is durably committed before the consumer exits.

Shutting the Poll Loop Down Cleanly

while(true) with a blocking poll() call has no way to exit on its own. The standard pattern is to call consumer.wakeup() from another thread (typically a JVM shutdown hook) — this interrupts an in-progress or subsequent poll() by throwing org.apache.kafka.common.errors.WakeupException on the polling thread, which you catch to break the loop and close the consumer cleanly:

Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    // ... process records ...
    consumer.commitAsync();
  }
} catch (WakeupException e) {
  // expected on shutdown, nothing to do
} finally {
  consumer.commitSync();
  consumer.close();
}

Skipping this means the JVM can only be stopped by killing the process outright, leaving the last batch’s offsets uncommitted and forcing a rebalance and reprocessing on the next start.

Gotcha: max.poll.interval.ms

If the code between two poll() calls takes too long — for example, slow downstream processing of a large batch — the broker assumes the consumer has died and kicks it out of the group, triggering a rebalance, even though the process is still running and hasn’t crashed. ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG ("max.poll.interval.ms", default 5 minutes) controls that timeout, and ConsumerConfig.MAX_POLL_RECORDS_CONFIG ("max.poll.records") controls how many records a single poll() returns — lowering it is often the simplest fix when per-record processing is slow enough to risk tripping the interval.