如何查询发布到Kafka主题的特定消息对应的分区索引/ID/编号

生产消息时实时获取（适配JBPM场景最优方案） #

Kafka 官方生产者客户端发送消息后返回的 RecordMetadata 对象原生包含分区索引信息，你可以在生产消息时直接获取后，和JBPM的流程实例、任务ID等业务标识绑定存储，后续无需二次查询。
Java 栈（JBPM默认技术栈）示例代码如下：

// 同步发送获取分区示例
ProducerRecord<String, String> record = new ProducerRecord<>("你的目标Topic名称", "消息Key", "消息内容");
try {
    RecordMetadata metadata = kafkaProducer.send(record).get();
    // 直接获取该消息对应的分区编号
    int partitionId = metadata.partition();
    // 同时可获取该分区下的消息偏移量，用于后续快速定位消息
    long offset = metadata.offset();
    // 此处自行实现分区、偏移量和JBPM业务字段的关联存储逻辑
} catch (InterruptedException | ExecutionException e) {
    // 消息发送异常处理逻辑
}

如果使用异步发送，直接在回调函数中取值即可：

kafkaProducer.send(record, (metadata, exception) -> {
    if (exception == null) {
        int partitionId = metadata.partition();
        long offset = metadata.offset();
        // 关联存储逻辑
    }
});

历史消息回溯查询方案 #

如果已经完成消息发送，没有提前存储分区信息，可以通过两种方式查询：

• 已知消息Key的场景：如果生产时没有自定义分区器，用Kafka默认分区策略直接计算即可，计算公式为 分区编号 = Utils.murmur2(消息Key的字节数组) % 目标Topic的总分区数，计算结果和实际分配的分区完全一致。
• 仅知道消息内容/Header特征的场景：遍历目标Topic的所有分区，拉取指定时间范围的消息匹配特征，匹配成功后从ConsumerRecord对象直接取分区编号：

// 1. 获取目标Topic的全部分区
List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor("目标Topic名称");
List<TopicPartition> allPartitions = partitionInfos.stream()
    .map(info -> new TopicPartition("目标Topic名称", info.partition()))
    .toList();
kafkaConsumer.assign(allPartitions);

// 2. 按需设置拉取起始位置，示例为从1小时前的偏移量开始拉取，减少无效查询
Map<TopicPartition, Long> timeQueryMap = allPartitions.stream()
    .collect(Collectors.toMap(tp -> tp, tp -> System.currentTimeMillis() - 3600 * 1000));
kafkaConsumer.offsetsForTimes(timeQueryMap).forEach((tp, offsetAndTs) -> {
    if (offsetAndTs != null) {
        kafkaConsumer.seek(tp, offsetAndTs.offset());
    } else {
        kafkaConsumer.seekToEnd(List.of(tp));
    }
});

// 3. 拉取消息匹配特征
while (true) {
    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        // 此处替换为你自己的消息匹配逻辑
        if (record.value().equals("待匹配的消息内容")) {
            int targetPartitionId = record.partition();
            long targetOffset = record.offset();
            // 匹配成功后处理逻辑，结束查询
            break;
        }
    }
}

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