Kafka重平衡监听器技术疑问：线程安全与示例代码问题

Great question—this is such an easy-to-overlook but critical detail when working with Kafka rebalance listeners, and you’re totally right to flag the thread safety concern here. Let’s break this down clearly:

核心问题：线程模型的差异 #

First, let’s recap how Kafka consumer threads operate:

• Your application’s main thread is the one calling poll() to fetch records, and it’s the same thread where you update that HashMap with processed offsets.
• Rebalance callback methods like onPartitionsRevoked() are triggered by internal Kafka client threads, not your application’s poll thread.

This means your HashMap is being accessed and modified by two entirely separate thread contexts. Without proper synchronization, you could run into messy race conditions:

• The poll thread might be updating an offset while the callback thread tries to read it, leading to partial or stale data.
• In worst-case scenarios, you could even hit a ConcurrentModificationException if the callback iterates over the map while it’s being modified.

为什么书中用普通HashMap？ #

The example in Kafka: The Definitive Guide uses a plain HashMap likely for simplicity—to keep the focus on how rebalance listeners work, not adding extra concurrency complexity. It’s a teaching tool, not production-ready code.

生产环境的解决方案 #

To fix this, you have a few solid, practical options:

1. 使用线程安全的集合 #

Replace the HashMap with ConcurrentHashMap from the java.util.concurrent package. This class is built for concurrent read/write operations, so you don’t have to handle synchronization manually:

private final Map<TopicPartition, OffsetAndMetadata> processedOffsets = new ConcurrentHashMap<>();

2. 显式同步访问 #

If you need to stick with a regular HashMap (unlikely, but possible), wrap all accesses to the map in synchronized blocks. For example:

// 在poll线程中更新偏移量
synchronized(processedOffsets) {
    processedOffsets.put(partition, new OffsetAndMetadata(offset + 1));
}

// 在onPartitionsRevoked回调中读取/提交偏移量
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    synchronized(processedOffsets) {
        consumer.commitSync(processedOffsets);
    }
}

3. 考虑持久化偏移量存储 #

For even better reliability, you might want to avoid in-memory storage entirely. Instead, persist processed offsets to a durable store (like a database) as you process records. Then, in the revoke callback, you can read the latest offsets from that store to commit. This eliminates in-memory concurrency issues and adds resilience if your application restarts.

额外注意事项 #

• Never share a single Kafka consumer instance across multiple application threads—consumers are not thread-safe. But the internal callback threads are managed by the client, so you have to account for their interactions with your application thread.
• When committing offsets in onPartitionsRevoked, make sure you only commit offsets for the partitions that are actually being revoked (not all partitions in the map). This avoids unnecessary commits and potential issues.

内容的提问来源于stack exchange，提问作者Justin Pihony