Apache Kafka is a distributed streaming platform designed to handle high-volume real-time data feeds. Kafka consumers read records from topic partitions, and managing their offsets - which track the position of the consumer in each partition - is crucial to ensure that the consumer processes each record exactly once, particularly in a read_committed transactional setting.

Understanding Offsets in Kafka

In Kafka, an offset is a sequential ID given to records as they are appended to a partition. The offset enables consumers to keep track of the records that have been read. When using Kafka, there are different configurations for consumers that define whether they read from the earliest offset, the latest, or resume from where they last stopped.

Consumer Groups and Offset Management

Consumers are often grouped into "consumer groups" for scalability and fault tolerance purposes. Kafka ensures that each partition is only consumed by one member of the group. Offsets are then committed by consumers to Kafka (specifically to a special Kafka topic named __consumer_offsets) so that if a consumer fails, another consumer from the same group can pick up reading from the last committed offset.

Read Committed Consumers

Read_committed is a configuration setting for Kafka consumers that influences how offsets are consumed and committed when working with transactional messages. In a transaction-enabled topic, producers can send messages in batches enclosed in transactions. These messages are visible to a read_committed consumer only after the transaction is committed.

Fetching the Last Committed Offset

To reliably fetch the last committed offset in a read_committed Kafka consumer environment, we use the following steps:

1. Consumer Configuration: Set the consumer in read_committed mode by setting the configuration isolation.level to read_committed.

1   Properties props = new Properties();
2   props.put("bootstrap.servers", "localhost:9092");
3   props.put("group.id", "test-group");
4   props.put("enable.auto.commit", "false");
5   props.put("isolation.level", "read_committed");
6   props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
7   props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
8   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

1. Subscribe to Topics: The consumer needs to be subscribed to the topic(s) from which it needs to read.

   consumer.subscribe(Arrays.asList("topic-name"));

1. Poll Messages: Continuously poll the Kafka broker to fetch new records. The consumer will see only those transactions which are committed if isolation.level is read_committed.

1   while (true) {
2       ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
3       for (ConsumerRecord<String, String> record : records)
4           System.out.println("offset = " + record.offset() + ", key = " + record.key() + ", value = " + record.value());
5   }

1. Fetching the Last Committed Offset: Use committed() method on the consumer to find the last committed offset for specific partitions.

1   Set<TopicPartition> partitions = consumer.assignment();
2   consumer.seekToEnd(partitions);
3   Map<TopicPartition, Long> endOffsets = new HashMap<>();
4   partitions.forEach(partition -> endOffsets.put(partition, consumer.position(partition) - 1));
5   
6   Map<TopicPartition, OffsetAndMetadata> committedOffsets = consumer.committed(partitions);
7   committedOffsets.forEach((partition, metadata) ->
8       System.out.println("Partition " + partition.partition() + " has last committed offset: " + metadata.offset())
9   );

This code effectively prints the last committed offset for each partition that the consumer is currently assigned. The combination of committed() and adjusting the position with seekToEnd() is crucial to determine the accurate position of the last committed transaction.

Summary Table

Here is a summary table of key aspects and method used in fetching the last committed offset:

By accurately monitoring and managing offsets, developers can ensure that their Kafka consumers are robust, maintain consistency, and handle failures gracefully, particularly in transaction-heavy environments.