是否有API可仅从Kafka获取元数据与消息头而不下载消息体？

针对Kafka主题中大Payload消息的元数据分析需求，以下是Python、Java、Go三种语言的实现方案，可避免不必要的Payload传输或内存加载：

Java 实现 #

方案1：低级API构造元数据专属Fetch请求 #

官方Kafka客户端未提供高层直接API，但可通过NetworkClient发送自定义FetchRequest，设置metadataOnly=true让服务端仅返回元数据（分区、Offset、时间戳、消息头等），完全跳过Payload传输：

import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.common.requests.FetchResponse;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.config.CommonClientConfigs;
import java.util.HashMap;
import java.util.Map;

public class KafkaMetaOnlyConsumer {
    public static void main(String[] args) {
        // 初始化客户端配置
        Map<String, Object> config = new HashMap<>();
        config.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        NetworkClient client = new NetworkClient(
            new org.apache.kafka.clients.DefaultMetadataUpdater(config, Time.SYSTEM),
            config,
            Time.SYSTEM,
            1024,
            1024,
            0,
            0,
            "",
            null,
            java.util.Collections.emptyList()
        );
        
        try {
            // 获取目标Broker节点
            client.awaitMetadataUpdate(1000);
            Node broker = client.leastLoadedNode();
            
            // 构造仅取元数据的Fetch请求
            FetchRequest request = FetchRequest.Builder
                .forConsumer("meta-only-group", 30000)
                .addFetch("your-topic", 0, 0, 1024) // 主题、分区、起始Offset、最大字节数
                .setMetadataOnly(true)
                .build();
            
            // 发送并处理响应
            client.send(request, broker, Time.SYSTEM);
            client.poll(1000, Time.SYSTEM);
            
            FetchResponse response = (FetchResponse) client.receive(broker, Time.SYSTEM).responseBody();
            response.responseData().forEach(fetchData -> {
                fetchData.records().forEach(record -> {
                    System.out.println("分区：" + fetchData.partition());
                    System.out.println("Offset：" + record.offset());
                    System.out.println("时间戳：" + record.timestamp());
                    System.out.println("消息头：" + record.headers());
                    // 无需调用record.value()，Payload未被传输
                });
            });
        } finally {
            client.close();
        }
    }
}

方案2：高层客户端内存优化 #

如果不想使用低级API，可在KafkaConsumer中绝不调用ConsumerRecord.value()，同时设置fetch.max.bytes为仅能容纳元数据和消息头的小值（如1024字节）。此方案虽仍会触发服务端尝试发送Payload，但超过阈值时会返回PARTITION_EOF，已传输的元数据可正常读取，能大幅减少内存占用。

Python 实现 #

方案1：高层消费者内存规避（简单易用） #

使用confluent-kafka库，只要不调用msg.value()，Payload就不会被反序列化加载到内存，仅占用网络传输资源（若需彻底规避网络传输，需用方案2）：

from confluent_kafka import Consumer, KafkaException, KafkaError

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'meta-only-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['your-topic'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise KafkaException(msg.error())
        
        # 仅提取元数据和消息头
        print(f"分区：{msg.partition()}")
        print(f"Offset：{msg.offset()}")
        print(f"时间戳：{msg.timestamp()[1]}")
        print(f"消息头：{[(h.key(), h.value()) for h in msg.headers()]}")
        # 不访问msg.value()，避免加载大Payload
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

方案2：底层协议请求（彻底规避Payload传输） #

通过confluent-kafka的底层Client构造FetchRequest并设置metadata_only=True：

from confluent_kafka import Client
from confluent_kafka.protocol import FetchRequest, FetchResponse

client = Client({'bootstrap.servers': 'localhost:9092'})
broker = client.brokers()[0]

# 构造仅取元数据的请求
fetch_req = FetchRequest(
    replica_id=-1,
    max_wait_time=1000,
    min_bytes=1,
    max_bytes=1024,
    metadata_only=True,
    topics=[('your-topic', [(0, 0, 1024)])]  # (分区, 起始Offset, 最大字节)
)

# 发送并解析响应
response = broker.send(fetch_req)
fetch_resp = FetchResponse(response)

for topic, partitions in fetch_resp.topics.items():
    for partition, data in partitions.items():
        for record in data.records:
            print(f"Offset: {record.offset}, Timestamp: {record.timestamp}, Headers: {record.headers}")

Go 实现 #

使用sarama库，直接构造FetchRequest并设置MetadataOnly=true，彻底跳过Payload传输：

package main

import (
    "log"
    "github.com/Shopify/sarama"
)

func main() {
    config := sarama.NewConfig()
    client, err := sarama.NewClient([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    // 获取主题分区信息
    partitions, err := client.Partitions("your-topic")
    if err != nil {
        log.Fatal(err)
    }

    // 构造仅取元数据的Fetch请求
    fetchRequest := &sarama.FetchRequest{
        Version:      sarama.MaxVersion,
        ReplicaID:    sarama.ConsumerReplicaID,
        MaxWaitTime:  1000,
        MinBytes:     1,
        MaxBytes:     1024,
        MetadataOnly: true, // 关键配置：仅返回元数据
    }

    // 添加目标分区与起始Offset
    for _, partition := range partitions {
        offset, err := client.GetOffset("your-topic", partition, sarama.OffsetOldest)
        if err != nil {
            log.Fatal(err)
        }
        fetchRequest.AddBlock("your-topic", partition, offset, 1024)
    }

    // 发送请求到Broker并解析响应
    broker, err := client.Leader("your-topic", partitions[0])
    if err != nil {
        log.Fatal(err)
    }
    response, err := broker.Fetch(fetchRequest)
    if err != nil {
        log.Fatal(err)
    }

    for topic, partitions := range response.Blocks {
        for partition, block := range partitions {
            for _, record := range block.Records {
                log.Printf("主题：%s，分区：%d，Offset：%d，时间戳：%d，消息头：%v",
                    topic, partition, record.Offset, record.Timestamp, record.Headers)
                // 无需访问record.Value，Payload未被传输
            }
        }
    }
}

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