使用assign()实现Kafka自定义分区分配未达预期求助

关键前提 #

使用 assign() 实现自定义分区分配时，不要让消费者加入共享的消费者组——要么不配置 group.id，要么给每个消费者设置唯一的 group.id。因为 assign() 是完全手动的分区分配逻辑，会绕过 Kafka 的消费者组自动分配机制，混用会导致预期外的分区重新分配。

正确代码实现（Java 示例） #

Consumer1 代码（消费分区1、3） #

Properties props = new Properties();
props.put("bootstrap.servers", "你的Kafka broker地址列表");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 禁用消费者组自动分配逻辑：不设置group.id 或设置唯一值
// props.put("group.id", "consumer1-unique-group");
// 首次消费时指定位移重置策略，确保能拿到历史消息
props.put("auto.offset.reset", "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 明确指定要消费的分区（TopicPartition 构造参数：主题名 + 分区号）
List<TopicPartition> targetPartitions = Arrays.asList(
    new TopicPartition("sampleTopic", 1),
    new TopicPartition("sampleTopic", 3)
);
consumer.assign(targetPartitions);

// 消费循环
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("Consumer1: 分区=%d, 位移=%d, Key=%s, Value=%s%n",
                record.partition(), record.offset(), record.key(), record.value());
        }
    }
} finally {
    consumer.close();
}

Consumer2/Consumer3 适配修改 #

• Consumer2 只需修改 targetPartitions 为 new TopicPartition("sampleTopic", 2) 和 new TopicPartition("sampleTopic", 5)
• Consumer3 改为 new TopicPartition("sampleTopic", 4) 和 new TopicPartition("sampleTopic", 6)

常见问题排查 #

1. 分区编号误解：Kafka 分区默认从0开始计数，先确认你的 sampleTopic 实际分区编号是否为1-6。用以下命令查看主题详情：

kafka-topics.sh --describe --topic sampleTopic --bootstrap-server 你的broker地址

如果实际分区是0-5，你需要调整代码中的分区号（比如原计划的分区1对应代码里的0）。
2. 混用 subscribe() 和 assign()：如果代码中先调用了 subscribe() 订阅主题，再调用 assign()，会覆盖订阅逻辑，但如果消费者已加入组，自动分配机制可能干扰手动分配，确保全程只用 assign()。
3. 重复分配分区：多个消费者不能 assign() 同一个分区，否则会导致重复消费，检查三个消费者的分区列表是否无重叠。
4. 位移未初始化：如果是首次消费，未设置 auto.offset.reset 会导致消费者无法获取初始位移，需配置为 earliest 或 latest。

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