Kafka Consumer Usage Example and Deep Dive into Offset Management, Rebalance, and Thread Safety

Usage Example

public static void main(String[] args) {
    Properties props = new Properties();
    String topic = "test";
    // auto-offset-commit
    String group = "test0";
    props.put("bootstrap.servers", "XXX:9092,XXX:9092");
    props.put("group.id", group);
    props.put("auto.offset.reset", "earliest");
    // 自动commit
    props.put("enable.auto.commit", "true");
    // 自动commit的间隔
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    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);
    // 可消费多个topic,组成一个list
    consumer.subscribe(Arrays.asList(topic));
    while (true) {
        ConsumerRecords
records = consumer.poll(100);
        for (ConsumerRecord
record : records) {
            System.out.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Offset Delivery Guarantees

At most once: messages may be lost but are never delivered twice.

At least once: messages are never lost but may be delivered more than once.

Exactly once: each message is delivered only once.

Exactly‑once Implementation

Producer can assign a globally unique ID to each message; consumer filters duplicates.

Consumer processing order vs. offset commit determines delivery semantics: Process then commit – may lead to at‑most‑once if a crash occurs before commit. Commit then process – may lead to at‑least‑once if a crash occurs after commit but before processing.

Solution: disable auto‑commit, combine offset commit and message processing in a single transaction (e.g., DB or Redis). On failure, rollback the transaction; on restart, retrieve the stored offset and seek to it.

During a rebalance, use a rebalance listener to commit offsets before the rebalance and seek to the stored offset after the rebalance.

Consumer Rebalance Triggers

Change in the number of members in the consumer group.

Change in the number of subscribed topics.

Change in the number of partitions for a subscribed topic.

Traditional Zookeeper‑based coordination can cause herd effects and split‑brain problems. Modern Kafka uses a GroupCoordinator that registers watches in Zookeeper only for metadata changes, then notifies consumers via the coordinator.

Partition Assignment Strategies

RoundRobinAssignor – iterates over all topic‑partitions and consumers in a round‑robin fashion.

RangeAssignor – assigns contiguous ranges of partitions to consumers; extra partitions are distributed to the first few consumers.

// Example calculation of partition distribution
// numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
// consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();

consumer 0: start: 0, length: 2, topic-partition: p0,p1;
consumer 1: start: 2, length: 2, topic-partition: p2,p3;
consumer 2: start: 4, length: 1, topic-partition: p4;
consumer 3: start: 5, length: 1, topic-partition: p5;
consumer 4: start: 6, length: 1, topic-partition: p6;

KafkaConsumer Analysis – Thread Safety

KafkaConsumer is not thread‑safe; callers must ensure single‑threaded access or use separate consumer instances per thread.

Common pattern: one thread pool pulls messages (each thread holds its own consumer) and places records into a queue; another pool processes the queue.

Request Handling Abstraction

Requests such as JoinGroup, SyncGroup, Heartbeat share a common send‑and‑listen pattern using RequestFuture and RequestFutureListener .

HeartbeatTask demonstrates how a periodic task sends a HeartbeatRequest and schedules the next execution based on success or failure.

public void run(final long now) {
    // send HeartbeatRequest
    RequestFuture
future = sendHeartbeatRequest();
    future.addListener(new RequestFutureListener
() {
        @Override public void onSuccess(Void value) {
            requestInFlight = false;
            long now = time.milliseconds();
            heartbeat.receiveHeartbeat(now);
            long next = now + heartbeat.timeToNextHeartbeat(now);
            client.schedule(HeartbeatTask.this, next);
        }
        @Override public void onFailure(RuntimeException e) {
            requestInFlight = false;
            client.schedule(HeartbeatTask.this, time.milliseconds() + retryBackoffMs);
        }
    });
}

Core poll() Method

public ConsumerRecords
poll(long timeout) {
    acquire();
    try {
        do {
            Map
>> records = pollOnce(remaining);
            if (!records.isEmpty()) {
                if (this.interceptors == null)
                    return new ConsumerRecords<>(records);
                else
                    return this.interceptors.onConsume(new ConsumerRecords<>(records));
            }
        } while (remaining > 0);
        return ConsumerRecords.empty();
    } finally {
        release();
    }
}

pollOnce() first ensures the GroupCoordinator is ready, performs rebalance if needed, updates fetch positions, executes delayed tasks (heartbeat, auto‑commit), fetches cached records, and finally sends fetch requests if the cache is empty.

Overall, the article provides a comprehensive guide to building a reliable Kafka consumer in Java, covering configuration, offset semantics, rebalance handling, partition assignment, thread‑safety, and the internal request/response flow.