Implementation:Fede1024 Rust rdkafka Metadata Wrappers

Knowledge Sources	Fede1024_Rust_rdkafka
Domains	Cluster_Administration, Metadata, FFI_Wrapper
Last Updated	2026-02-07 19:00 GMT

Overview

Safe Rust wrappers around librdkafka's cluster metadata C structures, providing typed access to broker information, topic details, and partition assignments.

Description

The metadata module defines four newtype wrapper structs: MetadataBroker, MetadataPartition, MetadataTopic, and Metadata. Each wraps the corresponding C struct from rdkafka-sys and provides safe accessor methods. MetadataBroker exposes broker ID, hostname, and port. MetadataPartition provides partition ID, leader broker, error state, replica IDs, and ISR (in-sync replica) lists. MetadataTopic gives topic name, partition list, and error state. The top-level Metadata struct owns a NativePtr<RDKafkaMetadata> that implements KafkaDrop for automatic deallocation via rd_kafka_metadata_destroy. All string fields are safely converted from C to Rust UTF-8.

Usage

Obtain a Metadata value by calling consumer.fetch_metadata() or client.fetch_metadata(). Use the accessor methods to inspect cluster topology, discover topics, check partition health, verify replication, and monitor leader distribution. This is the primary API for programmatic cluster inspection.

Code Reference

Source Location

• Repository: Fede1024_Rust_rdkafka
• File: src/metadata.rs
• Lines: 1-212

Signature

pub struct MetadataBroker(RDKafkaMetadataBroker);

impl MetadataBroker {
    pub fn id(&self) -> i32;
    pub fn host(&self) -> &str;
    pub fn port(&self) -> i32;
}

pub struct MetadataPartition(RDKafkaMetadataPartition);

impl MetadataPartition {
    pub fn id(&self) -> i32;
    pub fn leader(&self) -> i32;
    pub fn error(&self) -> Option<RDKafkaRespErr>;
    pub fn replicas(&self) -> &[i32];
    pub fn isr(&self) -> &[i32];
}

pub struct MetadataTopic(RDKafkaMetadataTopic);

impl MetadataTopic {
    pub fn name(&self) -> &str;
    pub fn partitions(&self) -> &[MetadataPartition];
    pub fn error(&self) -> Option<RDKafkaRespErr>;
}

pub struct Metadata(NativePtr<RDKafkaMetadata>);

impl Metadata {
    pub fn orig_broker_id(&self) -> i32;
    pub fn orig_broker_name(&self) -> &str;
    pub fn brokers(&self) -> &[MetadataBroker];
    pub fn topics(&self) -> &[MetadataTopic];
}

Import

use rdkafka::metadata::{Metadata, MetadataBroker, MetadataTopic, MetadataPartition};

I/O Contract

Inputs

Name	Type	Required	Description
Raw metadata pointer	*const RDKafkaMetadata	Yes	Obtained from rd_kafka_metadata() FFI call

Outputs

Name	Type	Description
brokers	&[MetadataBroker]	Slice of broker info (id, host, port)
topics	&[MetadataTopic]	Slice of topic info with nested partitions
orig_broker_id	i32	ID of the broker that responded to the metadata request
orig_broker_name	&str	Name of the responding broker

Usage Examples

Fetching and Inspecting Cluster Metadata

use rdkafka::consumer::{BaseConsumer, Consumer};
use rdkafka::config::ClientConfig;
use std::time::Duration;

let consumer: BaseConsumer = ClientConfig::new()
    .set("bootstrap.servers", "localhost:9092")
    .create()
    .expect("Consumer creation failed");

let metadata = consumer
    .fetch_metadata(None, Duration::from_secs(10))
    .expect("Failed to fetch metadata");

// Responding broker
println!("Metadata from broker: {} ({})",
    metadata.orig_broker_name(), metadata.orig_broker_id());

// List all brokers
for broker in metadata.brokers() {
    println!("  Broker {}: {}:{}", broker.id(), broker.host(), broker.port());
}

// List topics and partitions
for topic in metadata.topics() {
    println!("Topic: {} ({} partitions)", topic.name(), topic.partitions().len());
    for partition in topic.partitions() {
        println!("  Partition {}: leader={}, replicas={:?}, isr={:?}",
            partition.id(),
            partition.leader(),
            partition.replicas(),
            partition.isr());
        if let Some(err) = partition.error() {
            println!("    Error: {:?}", err);
        }
    }
}

Related Pages

• Environment:Fede1024_Rust_rdkafka_Rust_Librdkafka_Build_Environment