Implementation:Fede1024 Rust rdkafka AdminClient

Knowledge Sources	Fede1024_Rust_rdkafka librdkafka Admin API
Domains	Cluster_Administration, Async_Programming, Kafka_Operations
Last Updated	2026-02-07 19:00 GMT

Overview

Async Kafka admin client for programmatic cluster management, including topic creation/deletion, partition management, record deletion, and configuration inspection/alteration.

Description

The AdminClient<C> struct wraps a librdkafka producer-type Client and maintains a dedicated NativeQueue with a background polling thread. Each admin operation converts Rust types into native librdkafka C structures via unsafe FFI calls, attaches a oneshot::Sender as an opaque pointer on AdminOptions, and enqueues the request. The background thread polls the queue for completion events and sends the NativeEvent back through the channel. Each operation returns a custom Future type (e.g., CreateTopicsFuture, DeleteTopicsFuture) that awaits the receiver and parses the native event into Rust result types. Builder-pattern types (NewTopic, AdminOptions) handle parameter construction.

Usage

Use AdminClient when you need to programmatically manage Kafka cluster resources: creating topics before producing, deleting topics in integration tests, managing partitions, deleting records for GDPR compliance, or inspecting/altering broker configurations. Create via ClientConfig::create::<AdminClient<DefaultClientContext>>().

Code Reference

Source Location

• Repository: Fede1024_Rust_rdkafka
• File: src/admin.rs
• Lines: 1-1353

Signature

pub struct AdminClient<C: ClientContext> {
    client: Client<C>,
    queue: Arc<NativeQueue>,
    should_stop: Arc<AtomicBool>,
    handle: Option<JoinHandle<()>>,
}

impl<C: ClientContext> AdminClient<C> {
    pub fn create_topics<'a, I>(
        &self,
        topics: I,
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<TopicResult>>>
    where
        I: IntoIterator<Item = &'a NewTopic<'a>>;

    pub fn delete_topics(
        &self,
        topic_names: &[&str],
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<TopicResult>>>;

    pub fn delete_groups(
        &self,
        group_names: &[&str],
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<GroupResult>>>;

    pub fn create_partitions<'a, I>(
        &self,
        partitions: I,
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<TopicResult>>>
    where
        I: IntoIterator<Item = &'a NewPartitions<'a>>;

    pub fn delete_records(
        &self,
        offsets: &TopicPartitionList,
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<TopicPartitionList>>;

    pub fn describe_configs<'a, I>(
        &self,
        configs: I,
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<ConfigResourceResult>>>
    where
        I: IntoIterator<Item = &'a ResourceSpecifier<'a>>;

    pub fn alter_configs<'a, I>(
        &self,
        configs: I,
        opts: &AdminOptions,
    ) -> impl Future<Output = KafkaResult<Vec<AlterConfigsResult>>>
    where
        I: IntoIterator<Item = &'a AlterConfig<'a>>;

    pub fn inner(&self) -> &Client<C>;
}

pub struct AdminOptions {
    request_timeout: Option<Timeout>,
    operation_timeout: Option<Timeout>,
    validate_only: bool,
    broker_id: Option<i32>,
}

pub struct NewTopic<'a> {
    pub name: &'a str,
    pub num_partitions: i32,
    pub replication: TopicReplication<'a>,
    pub config: Vec<(&'a str, &'a str)>,
}

pub enum TopicReplication<'a> {
    Fixed(i32),
    Variable(PartitionAssignment<'a>),
}

pub type TopicResult = Result<String, (String, RDKafkaErrorCode)>;
pub type GroupResult = Result<String, (String, RDKafkaErrorCode)>;

Import

use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
use rdkafka::client::DefaultClientContext;
use rdkafka::config::ClientConfig;

I/O Contract

Inputs

Name	Type	Required	Description
ClientConfig	ClientConfig	Yes	Kafka connection settings (bootstrap.servers, etc.)
topics	&[NewTopic]	Per-method	Topic specifications for create_topics
topic_names	&[&str]	Per-method	Topic names for delete_topics
group_names	&[&str]	Per-method	Group names for delete_groups
opts	&AdminOptions	Yes	Request/operation timeouts and validation flags

Outputs

Name	Type	Description
TopicResult	Result<String, (String, RDKafkaErrorCode)>	Per-topic success (topic name) or failure (topic name + error)
GroupResult	Result<String, (String, RDKafkaErrorCode)>	Per-group success or failure
TopicPartitionList	TopicPartitionList	Post-deletion low-water marks for delete_records
ConfigResourceResult	Vec<ConfigEntry>	Configuration entries for describe_configs

Usage Examples

Creating Topics

use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
use rdkafka::client::DefaultClientContext;
use rdkafka::config::ClientConfig;

async fn create_topics() {
    // 1. Build admin client from config
    let admin: AdminClient<DefaultClientContext> = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .create()
        .expect("Admin client creation failed");

    // 2. Define topics to create
    let topics = [
        NewTopic::new("my-topic", 3, TopicReplication::Fixed(1))
            .set("cleanup.policy", "compact"),
        NewTopic::new("events", 6, TopicReplication::Fixed(3)),
    ];

    // 3. Execute with default options
    let opts = AdminOptions::new();
    let results = admin.create_topics(&topics, &opts).await;

    // 4. Check results
    match results {
        Ok(topic_results) => {
            for result in topic_results {
                match result {
                    Ok(name) => println!("Created topic: {}", name),
                    Err((name, err)) => eprintln!("Failed {}: {:?}", name, err),
                }
            }
        }
        Err(e) => eprintln!("Admin request failed: {}", e),
    }
}

Deleting Topics

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

async fn delete_topics(admin: &AdminClient<DefaultClientContext>) {
    let opts = AdminOptions::new()
        .request_timeout(Some(std::time::Duration::from_secs(5)));

    let results = admin
        .delete_topics(&["my-topic", "events"], &opts)
        .await
        .expect("Delete request failed");

    for result in results {
        match result {
            Ok(name) => println!("Deleted topic: {}", name),
            Err((name, err)) => eprintln!("Failed to delete {}: {:?}", name, err),
        }
    }
}

Related Pages

• Environment:Fede1024_Rust_rdkafka_Rust_Librdkafka_Build_Environment
• Environment:Fede1024_Rust_rdkafka_Kafka_Broker_Runtime