Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Fede1024 Rust rdkafka Metadata Example

From Leeroopedia


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

Overview

Example CLI application that fetches and displays Kafka cluster metadata including broker information, topic details, partition assignments, replica sets, ISR lists, and optionally watermark offsets.

Description

The metadata example creates a BaseConsumer (synchronous, non-streaming consumer variant) and calls consumer.fetch_metadata() with an optional topic filter and configurable timeout. The returned Metadata object is traversed to print cluster-level info, per-broker details (host, port, ID), and per-topic/partition details (leader, replicas, ISR). When the --offsets flag is provided, it additionally calls consumer.fetch_watermarks() for each partition to display low and high watermarks. Uses clap for CLI argument parsing.

Usage

Run this example to inspect a Kafka cluster's topology, discover topics, and check partition health. Useful for debugging replication issues, verifying topic configurations, and monitoring partition leadership distribution.

Code Reference

Source Location

Signature

fn print_metadata(
    brokers: &String,
    topic: Option<&String>,
    timeout: Duration,
    fetch_offsets: bool,
)

fn main()  // CLI entry point with clap argument parsing

Import

use rdkafka::consumer::{BaseConsumer, Consumer};
use rdkafka::config::ClientConfig;
use rdkafka::metadata::Metadata;

I/O Contract

Inputs

Name Type Required Description
brokers String (CLI arg) Yes Comma-separated broker list (e.g., "localhost:9092")
topic Option<String> (CLI arg) No Optional topic name to filter metadata
timeout Duration Yes Metadata fetch timeout (default 60s)
fetch_offsets bool (CLI flag) No Whether to also fetch watermark offsets

Outputs

Name Type Description
stdout Text Formatted cluster metadata: brokers, topics, partitions, replicas, ISR, watermarks

Usage Examples

Running the Metadata Example

# Fetch all cluster metadata
cargo run --example metadata -- --brokers localhost:9092

# Fetch metadata for a specific topic with offsets
cargo run --example metadata -- --brokers localhost:9092 --topic my-topic --offsets

Key Code Pattern: Fetching 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");

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

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

// Iterate topics and partitions
for topic in metadata.topics() {
    println!("Topic: {}", topic.name());
    for partition in topic.partitions() {
        println!("  Partition {}: leader={}, replicas={:?}, isr={:?}",
            partition.id(), partition.leader(),
            partition.replicas(), partition.isr());
    }
}

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment