Struct KafkaAdmin 

pub struct KafkaAdmin { /* private fields */ }

Client for Kafka admin operations.

KafkaAdmin is an async client for cluster, topic, group, config, feature, and SCRAM credential operations. Construct it either from an AdminConfig when you need explicit configuration, or through KafkaClient::admin for the builder-style API.

use kafkit_client::{AdminConfig, KafkaAdmin, NewTopic};

let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;

admin.create_topics([NewTopic::new("orders", 3, 1)]).await?;
let topics = admin.list_topics().await?;

Implementations

impl KafkaAdmin

pub async fn connect(config: AdminConfig) -> Result<Self>

Connects to Kafka and returns an admin client.

This performs a lightweight bootstrap connection so configuration, security settings, and broker reachability fail before the client is returned.

pub async fn create_topics<I>(&self, topics: I) -> Result<()>

where I: IntoIterator<Item = NewTopic>,

Creates one or more topics.

Existing topics are treated as success, matching the common create-if-missing setup flow. Passing an empty iterator is a no-op.

pub async fn delete_topics<I, S>(&self, topics: I) -> Result<()>

where I: IntoIterator<Item = S>, S: Into<String>,

Deletes one or more topics by name.

Passing an empty iterator is a no-op. Kafka must have topic deletion enabled for the broker to complete the operation.

pub async fn create_partitions<I, S>(&self, partitions: I) -> Result<()>

where I: IntoIterator<Item = (S, NewPartitions)>, S: Into<String>,

Increases partition counts for one or more topics.

Each item is a (topic_name, NewPartitions) pair. Kafka only supports increasing the total partition count; it cannot shrink a topic.

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

Lists topics known to the cluster.

The returned list is sorted by topic name.

pub async fn describe_topics<I, S>( &self, topics: I, ) -> Result<Vec<TopicDescription>>

where I: IntoIterator<Item = S>, S: Into<String>,

Describes selected topics.

The returned descriptions preserve the input topic order. Passing an empty iterator returns an empty vector.

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

Describes cluster ID, controller, and broker endpoints.

pub async fn describe_log_dirs<I>( &self, brokers: I, topics: Option<&[TopicPartition]>, ) -> Result<Vec<BrokerLogDirs>>

where I: IntoIterator<Item = BrokerDescription>,

Describes broker log directories and replica log sizes.

Kafka serves this API from each broker. Pass broker metadata returned by KafkaAdmin::describe_cluster. When topics is None, Kafka returns all topic-partitions hosted by each broker.

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

Lists consumer groups known to the cluster.

The returned list is sorted by group ID.

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

Alias for KafkaAdmin::list_groups.

pub async fn describe_groups<I, S>( &self, groups: I, ) -> Result<Vec<GroupDescription>>

where I: IntoIterator<Item = S>, S: Into<String>,

Describes groups, routing each group to the appropriate Kafka describe API.

Newer brokers report group type in ListGroups; this method uses that metadata to describe consumer, share, and classic groups. The returned descriptions preserve the input group order.

pub async fn describe_consumer_groups<I, S>( &self, groups: I, ) -> Result<Vec<ConsumerGroupDescription>>

where I: IntoIterator<Item = S>, S: Into<String>,

Describes KIP-848 consumer groups.

Use KafkaAdmin::describe_groups when the group type is not known.

pub async fn describe_classic_groups<I, S>( &self, groups: I, ) -> Result<Vec<ConsumerGroupDescription>>

where I: IntoIterator<Item = S>, S: Into<String>,

Describes classic consumer groups with Kafka’s legacy describe-groups API.

Use KafkaAdmin::describe_groups when the group type is not known.

pub async fn describe_share_groups<I, S>( &self, groups: I, ) -> Result<Vec<ConsumerGroupDescription>>

where I: IntoIterator<Item = S>, S: Into<String>,

Describes Kafka share groups.

Use KafkaAdmin::describe_groups when the group type is not known.

pub async fn delete_groups<I, S>(&self, groups: I) -> Result<()>

where I: IntoIterator<Item = S>, S: Into<String>,

Deletes one or more groups by ID.

Passing an empty iterator is a no-op.

pub async fn delete_consumer_groups<I, S>(&self, groups: I) -> Result<()>

where I: IntoIterator<Item = S>, S: Into<String>,

Alias for KafkaAdmin::delete_groups.

pub async fn describe_acls( &self, filter: AclBindingFilter, ) -> Result<Vec<AclBinding>>

Describes ACL bindings matching a filter.

The shape mirrors Kafka’s Admin#describeAcls(AclBindingFilter).

pub async fn create_acls<I>(&self, acls: I) -> Result<()>

where I: IntoIterator<Item = AclBinding>,

Creates ACL bindings.

The shape mirrors Kafka’s Admin#createAcls(Collection<AclBinding>).

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

where I: IntoIterator<Item = AclBindingFilter>,

Deletes ACL bindings matching each filter.

The shape mirrors Kafka’s Admin#deleteAcls(Collection<AclBindingFilter>).

pub async fn describe_configs<I>( &self, resources: I, ) -> Result<Vec<ConfigResourceConfig>>

where I: IntoIterator<Item = ConfigResource>,

Describes configuration entries for Kafka resources.

use kafkit_client::ConfigResource;

let configs = admin
    .describe_configs([ConfigResource::topic("orders")])
    .await?;

if let Some(retention) = configs[0].entry("retention.ms") {
    println!("retention: {:?}", retention.value);
}

pub async fn incremental_alter_configs<I>(&self, resources: I) -> Result<()>

where I: IntoIterator<Item = (ConfigResource, Vec<AlterConfigOp>)>,

Applies incremental config changes to Kafka resources.

use kafkit_client::{AlterConfigOp, ConfigResource};

admin
    .incremental_alter_configs([(
        ConfigResource::topic("orders"),
        vec![AlterConfigOp::set("retention.ms", "604800000")],
    )])
    .await?;

pub async fn upsert_scram_credential( &self, user: impl Into<String>, mechanism: SaslMechanism, password: impl AsRef<[u8]>, ) -> Result<()>

Creates or replaces a SCRAM credential for a user.

mechanism must be SaslMechanism::ScramSha256 or SaslMechanism::ScramSha512. Use KafkaAdmin::upsert_scram_credential_with_iterations to override the default iteration count.

pub async fn upsert_scram_credential_with_iterations( &self, user: impl Into<String>, mechanism: SaslMechanism, password: impl AsRef<[u8]>, iterations: i32, ) -> Result<()>

Creates or replaces a SCRAM credential with an explicit iteration count.

pub fn config(&self) -> &AdminConfig

Returns the configuration used by this admin client.

pub async fn finalized_feature_levels(&self) -> Result<Vec<BrokerFeatureLevel>>

Returns finalized broker feature levels visible from a bootstrap broker.

pub async fn update_features<I>(&self, updates: I) -> Result<()>

where I: IntoIterator<Item = FeatureUpdate>,

Updates finalized broker feature levels.

Feature updates are cluster-level operations. Consider calling KafkaAdmin::validate_feature_updates first when supported by the broker.

pub async fn validate_feature_updates<I>(&self, updates: I) -> Result<()>

where I: IntoIterator<Item = FeatureUpdate>,

Validates finalized broker feature updates without applying them.

Trait Implementations

impl Clone for KafkaAdmin

fn clone(&self) -> KafkaAdmin

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for KafkaAdmin

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.