Kafka跨多Topic消费时同分区数据未分配至同一消费者问题

当前配置的RangeAssignor以单个Topic为独立单位执行分区区间划分，分配逻辑不会跨Topic对齐分区编号，多Topic订阅场景下天然会出现同编号分区被分散到不同消费者的问题，和观测到的异常分配结果完全吻合。

方案1：自定义同分区号绑定分配策略（优先推荐，支持自动再均衡） #

实现Kafka提供的AbstractPartitionAssignor抽象类，核心逻辑是先将所有Topic的分区按分区编号分组，再把同编号的全部分区作为一个整体分配给同一个消费者，既满足同编号分区共置的需求，又保留Kafka原生的自动再均衡、故障转移、扩缩容自动适配能力。
实现代码示例：

public class SamePartitionIdAssignor extends AbstractPartitionAssignor {
    @Override
    public String name() {
        return "same-partition-id-assignor";
    }

    @Override
    public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic, Map<String, Subscription> subscriptions) {
        // 初始化消费者分配结果容器，消费者按ID字典序排序保证分配一致性
        List<String> consumers = new ArrayList<>(subscriptions.keySet());
        Collections.sort(consumers);
        Map<String, List<TopicPartition>> assignment = new HashMap<>();
        consumers.forEach(c -> assignment.put(c, new ArrayList<>()));

        // 按分区编号聚合所有Topic的对应分区
        Map<Integer, List<TopicPartition>> partitionIdGroup = new TreeMap<>();
        partitionsPerTopic.forEach((topic, partitionCount) -> {
            for (int partitionId = 0; partitionId < partitionCount; partitionId++) {
                partitionIdGroup.computeIfAbsent(partitionId, k -> new ArrayList<>())
                        .add(new TopicPartition(topic, partitionId));
            }
        });

        List<Integer> allPartitionIds = new ArrayList<>(partitionIdGroup.keySet());
        int consumerCount = consumers.size();
        int basePartitionPerConsumer = allPartitionIds.size() / consumerCount;
        int remainder = allPartitionIds.size() % consumerCount;
        int assignOffset = 0;

        // 按顺序给每个消费者分配连续的分区编号组，保证同编号分区落到同一消费者
        for (String consumer : consumers) {
            int currentAssignCount = basePartitionPerConsumer + (remainder-- > 0 ? 1 : 0);
            List<TopicPartition> currentAssignPartitions = new ArrayList<>();
            for (int i = assignOffset; i < assignOffset + currentAssignCount; i++) {
                currentAssignPartitions.addAll(partitionIdGroup.get(allPartitionIds.get(i)));
            }
            assignment.get(consumer).addAll(currentAssignPartitions);
            assignOffset += currentAssignCount;
        }
        return assignment;
    }
}

替换消费者工厂中的分配策略配置即可生效：

props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, SamePartitionIdAssignor.class.getName());

该方案适配任意消费者数量、任意等分区数的多Topic场景：比如2个消费者时，每个消费者绑定10个连续分区编号对应的所有Topic分区；扩到4个消费者时，每个消费者绑定5个连续分区编号对应的所有Topic分区，始终满足同编号分区共置要求。

方案2：手动绑定分区（适合实例固定无扩缩容的场景） #

如果消费组实例数量长期固定、不需要自动故障转移，可以直接调用Kafka消费者的assign()方法手动给每个实例绑定固定分区，完全绕开自动分配逻辑。
示例代码（以2个消费者为例）：

List<TopicPartition> targetPartitions = new ArrayList<>();
List<String> allTopics = Arrays.asList("Topic1", "Topic2", "Topic3", "Topic4");
// 消费者1绑定所有Topic的0-9分区
for (String topic : allTopics) {
    for (int p = 0; p < 10; p++) {
        targetPartitions.add(new TopicPartition(topic, p));
    }
}
// 消费者2绑定所有Topic的10-19分区，替换循环范围即可
// for (int p = 10; p < 20; p++) { ... }
consumer.assign(targetPartitions);

该方案逻辑简单无兼容问题，但消费者宕机、扩缩容时需要人工调整分区绑定关系，运维成本较高。

• 消费组内所有消费者必须使用完全一致的分区分配策略，否则会触发反复重平衡，导致分配结果混乱
• 自定义分配策略类需要打包到所有消费者实例的classpath中，避免启动时类加载失败

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