Kafka Parallel Consumer多进程无法拆分任务问题求助

问题描述 #

我使用Confluent Parallel Consumer实现向不同数据存储的快速写入，本地Docker环境运行正常。但在多主机部署多个同Group ID的消费者后，发现仅一个节点（进程）在消费数据。当前读取的Topic有24个分区，部署了3个节点，预期Kafka会将任务拆分到各节点，相关代码如下：

消费者构建代码 #

fun buildConsumer(config: KafkaConsumerConfig): KafkaConsumer<String, JsonObject> {
    val props = Properties()
    props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = config.kafkaBootstrapServers
    props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
    props[ConsumerConfig.GROUP_ID_CONFIG] = "myGroup"
    // Auto commit must be false in parallel consumer
    props[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = false
    props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java.name
    props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = JsonObjectDeSerializer::class.java.name
    val consumer = KafkaConsumer<String, JsonObject>(props)
    
    return consumer
}

private fun createReactParallelConsumer(): ReactorProcessor<String, JsonObject> {
    val options = ParallelConsumerOptions.builder<String, JsonObject>()
        .ordering(ParallelConsumerOptions.ProcessingOrder.KEY)
        .maxConcurrency(10)
        .batchSize(1)
        .consumer(buildConsumer(kafkaConsumerConfig))
        .build()
    return ReactorProcessor(options)
}

主消费代码 #

pConsumer = createReactParallelConsumer()
pConsumer.subscribe(UniLists.of(kafkaConsumerConfig.kafkaTopic))
pConsumer.react { context ->
    batchProcessor.processBatch(context)
}

解决建议 #

• 检查Parallel Consumer的分区分配策略
  确保Parallel Consumer使用Kafka标准的分区分配策略，避免因自定义策略导致分区无法正常分配。可以在ParallelConsumerOptions中显式指定分配策略：

  .partitionAssignor(PartitionAssignor.ROUND_ROBIN) // 或RANGE，根据业务需求选择
  

  同时确认消费者的pollTimeout设置合理，避免因超时过短导致无法参与重平衡。

• 避免重复的group.instance.id配置
  如果代码中设置了group.instance.id，必须保证每个节点的该配置唯一——重复的实例ID会让Kafka将多个节点识别为同一个消费者实例，仅分配分区给其中一个。若未手动设置该属性，Kafka会自动生成唯一ID，此时需确保节点间网络连通，组协调器能正确识别不同实例。

• 调整重平衡相关配置
  若节点无法触发重平衡，可调整消费者的会话超时和心跳间隔配置，确保节点能被组协调器正常感知：

  props[ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG] = "30000"
  props[ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG] = "10000"
  

  同时查看Kafka broker日志，搜索"rebalance"关键词，确认重平衡事件是否触发、各节点是否参与。

• 验证处理顺序配置的影响
  当前使用ProcessingOrder.KEY模式，虽保证同Key消息顺序，但如果所有消息Key哈希后集中在少数分区，可能导致消费集中在单个节点。可临时改为ProcessingOrder.UNORDERED测试，排除顺序逻辑导致的分区独占问题。

• 确认订阅方式正确性
  确保所有节点均使用subscribe方法订阅Topic，而非assign手动分配分区——手动分配会绕过Kafka的组管理机制，导致分区无法自动拆分到多个节点。

• 排查节点日志与连接状态
  查看每个消费者节点的日志，检查是否存在连接Kafka失败、重平衡报错等信息，确认节点已成功加入消费组。

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