Struct AdminClient 

pub struct AdminClient { /* private fields */ }

Kafka admin client for cluster administration.

Implementations

impl AdminClient

pub fn builder() -> AdminClientBuilder

Create a new admin client builder.

pub async fn create_topics( &self, topics: Vec<NewTopic>, timeout: Duration, ) -> Result<Vec<CreateTopicResult>>

Create topics.

pub async fn delete_topics( &self, topics: Vec<String>, timeout: Duration, ) -> Result<Vec<DeleteTopicResult>>

Delete topics.

pub async fn create_partitions( &self, topic: impl Into<String>, new_total_count: i32, timeout: Duration, ) -> Result<CreatePartitionsResult>

Increase the number of partitions for a topic.

Note: Partition count can only be increased, never decreased.

pub async fn describe_topic_config( &self, topic: &str, ) -> Result<Vec<ConfigEntry>>

Describe configuration for a topic.

pub async fn describe_broker_config( &self, broker_id: i32, ) -> Result<Vec<ConfigEntry>>

Describe configuration for a broker.

pub async fn alter_topic_config( &self, topic: &str, configs: HashMap<String, String>, ) -> Result<AlterConfigResult>

Alter configuration for a topic.

Note: This replaces all dynamic configs. To modify a single config, first describe the topic config and then set all desired values.

pub async fn list_topics(&self) -> Result<Vec<String>>

List all topics.

pub async fn describe_topics(&self, topics: &[String]) -> Result<Vec<TopicInfo>>

Describe topics.

pub async fn describe_cluster(&self) -> Result<ClusterDescription>

Describe the cluster.

pub async fn partition_count(&self, topic: &str) -> Result<Option<usize>>

Get partition count for a topic.

pub fn client_id(&self) -> &str

Get the client ID.

pub fn request_timeout(&self) -> Duration

Get the request timeout.

pub async fn describe_acls( &self, resource_type: AclResourceType, resource_name: Option<&str>, pattern_type: AclPatternType, principal: Option<&str>, host: Option<&str>, operation: AclOperation, permission_type: AclPermissionType, ) -> Result<DescribeAclsResult>

Describe ACLs matching a filter.

Arguments

• resource_type - Type of resource (Topic, Group, Cluster, etc.)
• resource_name - Name of the resource (use None to match any)
• pattern_type - Pattern type (Literal, Prefixed, Any)
• principal - Principal (use None to match any)
• host - Host (use None to match any)
• operation - Operation (use Any to match all)
• permission_type - Permission type (use Any to match all)

Example

// Describe all ACLs for a specific topic
let result = admin.describe_acls(
    AclResourceType::Topic,
    Some("my-topic"),
    AclPatternType::Literal,
    None,
    None,
    AclOperation::Any,
    AclPermissionType::Any,
).await?;

pub async fn describe_acls_with_filter( &self, filter: AclFilter, ) -> Result<DescribeAclsResult>

Describe ACLs matching a filter.

This is the preferred method for describing ACLs as it uses a structured filter object.

Example

// Describe all ACLs for a specific topic
let filter = AclFilter::for_resource(AclResourceType::Topic, "my-topic");
let result = admin.describe_acls_with_filter(filter).await?;

pub async fn create_acls( &self, acls: Vec<AclBinding>, ) -> Result<CreateAclsResult>

Create ACLs.

Arguments

• acls - List of ACL bindings to create

Example

let acl = AclBinding::allow_read_topic("my-topic", "User:alice");
admin.create_acls(vec![acl]).await?;

pub async fn delete_acls( &self, filters: Vec<AclBindingFilter>, ) -> Result<DeleteAclsResult>

Delete ACLs matching the specified filters.

Arguments

• filters - List of ACL binding filters to match for deletion

Example

// Delete all ACLs for a specific topic
let filter = AclBindingFilter {
    resource_type: AclResourceType::Topic,
    resource_name: Some("my-topic".to_string()),
    pattern_type: AclPatternType::Literal,
    principal: None,
    host: None,
    operation: AclOperation::Any,
    permission_type: AclPermissionType::Any,
};
admin.delete_acls(vec![filter]).await?;

pub async fn describe_groups( &self, group_ids: Vec<String>, ) -> Result<Vec<ConsumerGroupDescription>>

Describe consumer groups.

Returns detailed information about each group including state, members, and partition assignments.

Example

let groups = admin.describe_groups(vec!["my-group".to_string()]).await?;
for group in &groups {
    println!("{}: state={}, members={}", group.group_id, group.state, group.members.len());
}

pub async fn list_consumer_groups(&self) -> Result<Vec<ConsumerGroupListing>>

List all consumer groups on the cluster.

Returns a list of all consumer groups with their protocol types.

Example

let groups = admin.list_consumer_groups().await?;
for group in &groups {
    println!("{} ({})", group.group_id, group.protocol_type);
}

pub async fn delete_records( &self, offsets: HashMap<(String, i32), i64>, timeout: Duration, ) -> Result<Vec<DeleteRecordResult>>

Delete records from topic partitions before the specified offsets.

Records with offsets less than the specified offset for each partition will be marked for deletion. This adjusts the log start offset.

Arguments

• offsets - Map of (topic, partition) to the offset before which to delete
• timeout - Operation timeout

Example

use std::collections::HashMap;
let mut offsets = HashMap::new();
offsets.insert(("my-topic".to_string(), 0), 100i64);
let results = admin.delete_records(offsets, Duration::from_secs(30)).await?;

pub async fn offset_for_leader_epoch( &self, partitions: Vec<(String, i32, i32)>, ) -> Result<Vec<LeaderEpochResult>>

Get the end offset for each partition at the given leader epoch.

This is used to detect log truncation after a leader change. For each topic-partition, the broker returns the end offset for the requested leader epoch. If the epoch is no longer valid, the broker returns the epoch and offset where the log was truncated.

Arguments

• partitions - List of (topic, partition, leader_epoch) tuples

Example

let results = admin.offset_for_leader_epoch(
    vec![("my-topic".to_string(), 0, 5)]
).await?;
for r in &results {
    println!("{}:{} epoch={} end_offset={}", r.topic, r.partition, r.leader_epoch, r.end_offset);
}

pub fn pool(&self) -> &Arc<ConnectionPool>

Get access to the connection pool.