rust-rdkafka消费者组成员分配格式及消费主题查询方法咨询

关于assignment()返回的二进制格式 #

你看到的&[u8]是Kafka协议原生的GroupAssignment二进制结构，属于标准格式，并非异常。rust-rdkafka基于librdkafka实现，后者直接返回了Kafka协议的原始字节流，而rust-rdkafka目前未对该字段做上层解析封装。

这个结构的具体规则对应Kafka官方协议定义：

• 每个分配项依次包含：
  1. 主题名称长度（2字节大端序无符号整数）
  2. 主题名称的UTF-8字节内容
  3. 分区列表长度（4字节大端序无符号整数）
  4. 每个分区的编号（4字节大端序整数）
     你观察到的“buffer”实际是这些长度字段和分区数据的字节，并非无效内容。

手动解析二进制数据的示例 #

如果需要解析这个字节数组，可以参考以下代码：

use std::io::{Cursor, Read};

fn parse_assignment(assignment: &[u8]) -> Vec<(String, Vec<i32>)> {
    let mut cursor = Cursor::new(assignment);
    let mut result = Vec::new();

    while cursor.position() < assignment.len() as u64 {
        // 读取主题长度
        let mut topic_len_buf = [0u8; 2];
        cursor.read_exact(&mut topic_len_buf).unwrap();
        let topic_len = u16::from_be_bytes(topic_len_buf) as usize;

        // 读取主题名称
        let mut topic_name_buf = vec![0u8; topic_len];
        cursor.read_exact(&mut topic_name_buf).unwrap();
        let topic_name = String::from_utf8(topic_name_buf).unwrap();

        // 读取分区数量
        let mut partition_count_buf = [0u8; 4];
        cursor.read_exact(&mut partition_count_buf).unwrap();
        let partition_count = u32::from_be_bytes(partition_count_buf) as usize;

        // 读取所有分区号
        let mut partitions = Vec::with_capacity(partition_count);
        for _ in 0..partition_count {
            let mut partition_buf = [0u8; 4];
            cursor.read_exact(&mut partition_buf).unwrap();
            let partition = i32::from_be_bytes(partition_buf);
            partitions.push(partition);
        }

        result.push((topic_name, partitions));
    }

    result
}

查询消费者组主题的更优方法 #

推荐使用rust-rdkafka的AdminClient提供的describe_groups API，它返回的结构已做封装，无需手动解析二进制：

use rdkafka::admin::{AdminClient, AdminOptions};
use rdkafka::ClientConfig;
use std::time::Duration;

async fn fetch_group_assigned_topics(brokers: &str, group_id: &str) -> Vec<String> {
    let admin_client = AdminClient::new(ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .create()
        .unwrap());

    let options = AdminOptions::new().request_timeout(Duration::from_secs(5));
    let group_descriptions = admin_client.describe_groups(&[group_id], &options).await.unwrap();

    let mut topics = Vec::new();
    if let Some(group_desc) = group_descriptions.get(group_id) {
        for member in group_desc.members() {
            if let Some(assignment) = member.assignment() {
                assignment.elements().for_each(|tp| {
                    topics.push(tp.topic().to_string());
                });
            }
        }
    }

    // 去重避免重复主题
    topics.dedup();
    topics
}

这个方法是专为集群管理场景设计的，返回的TopicPartitionList可以直接提取主题和分区信息，比fetch_group_list更易用。

另外，如果你不需要在代码中实现，也可以使用Kafka官方命令行工具：

kafka-consumer-groups.sh --bootstrap-server <你的broker地址> --describe --group <消费者组ID>

输出中的TOPIC列即为该组正在消费的主题。

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