如何使用需指定topic与partition的Rust rdkafka API？

1. 用Consumer拉取Topic元数据 #

这是最直接的方式，利用已有的Consumer实例获取目标Topic的完整元数据，从中提取分区索引和数量：

use rdkafka::consumer::Consumer;

// 假设你已经完成consumer的初始化
let timeout = std::time::Duration::from_secs(5);
match consumer.fetch_metadata(Some(topic), timeout) {
    Ok(metadata) => {
        if let Some(topic_meta) = metadata.topics().iter().find(|t| t.name() == topic) {
            let partition_indices: Vec<i32> = topic_meta.partitions().iter().map(|p| p.id()).collect();
            let partition_count = partition_indices.len();
            // 这里可以把分区信息保存下来供后续API使用
            println!("Topic {} 共{}个分区，索引: {:?}", topic, partition_count, partition_indices);
        } else {
            eprintln!("未找到目标Topic: {}", topic);
        }
    }
    Err(e) => eprintln!("获取元数据失败: {}", e),
}

2. 用AdminClient查询Topic详情 #

如果不需要消费能力，单独用AdminClient查询更轻量化：

use rdkafka::admin::{AdminClient, AdminOptions};
use rdkafka::config::ClientConfig;

// 初始化AdminClient，配置和Consumer一致即可
let admin_client = ClientConfig::new()
    .set("bootstrap.servers", "your-kafka-brokers")
    .create::<AdminClient<_>>()
    .unwrap();

let timeout = std::time::Duration::from_secs(5);
let admin_opts = AdminOptions::new().request_timeout(timeout);

match admin_client.describe_topics(&[topic], &admin_opts).await {
    Ok(topics) => {
        for topic_meta in topics {
            if topic_meta.name == topic {
                let partition_indices: Vec<i32> = topic_meta.partitions.iter().map(|p| p.id).collect();
                println!("分区索引列表: {:?}", partition_indices);
            }
        }
    }
    Err(e) => eprintln!("查询Topic失败: {}", e),
}

3. 基于分区索引处理历史消息 #

拿到合法的分区索引后，就可以逐个分区处理历史消息，适配你提到的组件重启后按分区读取的场景：

// 假设已经获取到partition_indices列表
let timeout = std::time::Duration::from_secs(5);
for &partition in &partition_indices {
    // 获取分区的水位线，确定历史消息的起始位置
    if let Ok((low_watermark, _)) = consumer.fetch_watermarks(topic, partition, timeout) {
        // 定位到分区的起始位置开始消费
        if let Err(e) = consumer.seek(topic, partition, low_watermark, timeout) {
            eprintln!("定位分区{}失败: {}", partition, e);
            continue;
        }
        // 这里添加你的历史消息消费逻辑
    } else {
        eprintln!("获取分区{}水位线失败", partition);
    }
}

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