Struct AdminClient 

pub struct AdminClient { /* private fields */ }

Short-lived admin client targeting one cluster’s controller. Optionally negotiates TLS/SASL via AdminClient::connect_secured.

Implementations

impl AdminClient

pub async fn describe_configs( &mut self, topics: &[&str], ) -> Result<Vec<TopicConfigOverrides>, AdminError>

pub async fn incremental_alter_configs( &mut self, ops: &[IncrementalAlterOp], ) -> Result<Vec<AlterConfigsOutcome>, AdminError>

impl AdminClient

pub async fn create_delegation_token_as_owner( &mut self, owner_principal_name: &str, renewers: &[String], max_lifetime_ms: i64, ) -> Result<CreateDelegationTokenResponse, AdminError>

KIP-48 act-as create: mint a delegation token whose owner is owner_principal_name (type always "User").

The caller MUST be a broker super-user (per broker semantics) for the act-as path to take effect; non-super callers get DELEGATION_TOKEN_AUTHORIZATION_FAILED (65). The full response is returned so callers can pluck out token_id, hmac, and the issue/expiry/max timestamps for downstream persistence.

renewers items are "User:bob" form; entries without a : are interpreted with type "User".

pub async fn renew_delegation_token( &mut self, hmac: &[u8], ) -> Result<i64, AdminError>

KIP-48 renew: bump the token’s expiry_timestamp_ms capped by max_timestamp_ms. renew_period_ms = -1 tells the broker to use its configured default. Returns the new expiry.

pub async fn expire_delegation_token( &mut self, hmac: &[u8], ) -> Result<(), AdminError>

KIP-48 expire: tombstone the token immediately (expiry_time_period_ms = -1).

pub async fn describe_delegation_tokens_owned_by( &mut self, owner_principal: &str, ) -> Result<Vec<DelegationToken>, AdminError>

KIP-48 describe filtered to a single owner. owner_principal is a canonical "Type:Name" string (e.g. "User:alice"); entries without a : default to type "User".

impl AdminClient

pub async fn alter_replica_log_dirs( &mut self, assignments: &BTreeMap<String, Vec<(String, Vec<i32>)>>, ) -> Result<Vec<AlterReplicaLogDirOutcome>, AdminError>

AlterReplicaLogDirs (KIP-113): move replicas between local log.dirs on this broker.

assignments maps each target absolute directory path to the (topic, [partition]) pairs that should be moved into it.

pub async fn describe_log_dirs( &mut self, filter: Option<&BTreeMap<String, Vec<i32>>>, ) -> Result<Vec<LogDirInfo>, AdminError>

DescribeLogDirs (KIP-113): list every configured log.dir on this broker, with the partitions each holds (current and in-progress future logs). Pass None to fetch all partitions or Some with topic → partitions filter (empty inner vec means all partitions of that topic).

impl AdminClient

pub async fn describe_user_quotas( &mut self, username: &str, ) -> Result<UserQuotaConfig, AdminError>

Read the broker’s current client-quota config for the named user. Filters strictly on the single-component entity [("user", Some(username))]; broker entries whose entity also carries a client-id axis do not match (matches Kafka admin-tool strict-component semantics).

pub async fn alter_user_quotas( &mut self, username: &str, ops: &[QuotaOp], validate_only: bool, ) -> Result<Option<KafkaError>, AdminError>

Apply ops against the (user) entity. Returns the per-entry KafkaError surfaced by the broker, or None on success.

validate_only mirrors the wire flag — when true the broker runs validation but writes no metadata record.

impl AdminClient

pub async fn metadata( &mut self, topics: &[&str], ) -> Result<TopicMetadata, AdminError>

Metadata for the named topics. Pass an empty slice to fetch all topics, per Kafka semantics.

pub async fn create_topics( &mut self, specs: &[CreateTopicSpec], timeout_ms: i32, ) -> Result<Vec<CreateTopicOutcome>, AdminError>

pub async fn delete_topics( &mut self, names: &[&str], timeout_ms: i32, ) -> Result<Vec<DeleteTopicOutcome>, AdminError>

pub async fn create_partitions( &mut self, ops: &[CreatePartitionsOp], timeout_ms: i32, ) -> Result<Vec<CreatePartitionsOutcome>, AdminError>

impl AdminClient

pub async fn alter_user_scram_credentials_sha512( &mut self, upsertions: &[ScramUpsertion], deletions: &[ScramDeletion], ) -> Result<Vec<ScramUserOutcome>, AdminError>

Upsert and/or delete SCRAM-SHA-512 credentials in a single call.

upsertions carry plaintext passwords — the function generates a fresh 16-byte salt per row and computes the KIP-554 wire salted_password (PBKDF2-HMAC-SHA-512) client-side via crabka_security::pbkdf2_salted. The broker never sees the raw password.

pub async fn alter_user_scram_credentials_sha256( &mut self, upsertions: &[ScramUpsertion], deletions: &[ScramDeletion], ) -> Result<Vec<ScramUserOutcome>, AdminError>

SCRAM-SHA-256 sibling of Self::alter_user_scram_credentials_sha512. Iteration counts, salt generation, and salted-password derivation are identical to the SHA-512 path; only the mechanism wire byte and HMAC algorithm differ.

Errors

Same as _sha512: returns AdminError::Protocol when the system RNG fails, otherwise propagates the broker’s per-username outcome rows.

pub async fn describe_acls( &mut self, filter: &AclEntryFilter, ) -> Result<Vec<AclEntry>, AdminError>

List ACLs matching filter. The broker’s response is resource-grouped on the wire (one block per (resource_type, resource_name, pattern_type)); we flatten back into AclEntry rows for diffing.

pub async fn create_acls( &mut self, creations: &[AclEntry], ) -> Result<Vec<CreateAclOutcome>, AdminError>

Create the supplied ACLs.

pub async fn delete_acls( &mut self, filters: &[AclEntryFilter], ) -> Result<Vec<DeleteAclFilterOutcome>, AdminError>

Delete every ACL matching any of filters. Each filter’s response surfaces the matched ACL set so callers can confirm the deletion converged on the expected rows.

impl AdminClient

pub async fn connect_secured( bootstrap_addrs: &[String], security: Option<ClientSecurity>, ) -> Result<Self, AdminError>

Connect, applying optional client security. None = plaintext (identical to AdminClient::connect).

Errors

Returns AdminError::Connect { tried } if no bootstrap address accepted the (optionally secured) connection.

pub async fn connect(bootstrap_addrs: &[String]) -> Result<Self, AdminError>

Try each bootstrap address in order. Each entry is host:port; DNS is resolved via tokio::net::lookup_host. First successful connect wins. Returns AdminError::Connect { tried } if none responded. Plaintext; see AdminClient::connect_secured.

Trait Implementations

impl AdminClientLike for AdminClient

fn metadata<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, topics: &'life1 [&'life2 str], ) -> Pin<Box<dyn Future<Output = Result<TopicMetadata, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn create_topics<'life0, 'life1, 'async_trait>( &'life0 mut self, specs: &'life1 [CreateTopicSpec], timeout_ms: i32, ) -> Pin<Box<dyn Future<Output = Result<Vec<CreateTopicOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn delete_topics<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, names: &'life1 [&'life2 str], timeout_ms: i32, ) -> Pin<Box<dyn Future<Output = Result<Vec<DeleteTopicOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn create_partitions<'life0, 'life1, 'async_trait>( &'life0 mut self, ops: &'life1 [CreatePartitionsOp], timeout_ms: i32, ) -> Pin<Box<dyn Future<Output = Result<Vec<CreatePartitionsOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn describe_configs<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, topics: &'life1 [&'life2 str], ) -> Pin<Box<dyn Future<Output = Result<Vec<TopicConfigOverrides>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn incremental_alter_configs<'life0, 'life1, 'async_trait>( &'life0 mut self, ops: &'life1 [IncrementalAlterOp], ) -> Pin<Box<dyn Future<Output = Result<Vec<AlterConfigsOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn alter_user_scram_credentials_sha512<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, upsertions: &'life1 [ScramUpsertion], deletions: &'life2 [ScramDeletion], ) -> Pin<Box<dyn Future<Output = Result<Vec<ScramUserOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn alter_user_scram_credentials_sha256<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, upsertions: &'life1 [ScramUpsertion], deletions: &'life2 [ScramDeletion], ) -> Pin<Box<dyn Future<Output = Result<Vec<ScramUserOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

SCRAM-SHA-256 sibling of Self::alter_user_scram_credentials_sha512. The operator calls this when a KafkaUser.spec.authentication.type == scram-sha-256.

fn describe_acls<'life0, 'life1, 'async_trait>( &'life0 mut self, filter: &'life1 AclEntryFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<AclEntry>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn create_acls<'life0, 'life1, 'async_trait>( &'life0 mut self, creations: &'life1 [AclEntry], ) -> Pin<Box<dyn Future<Output = Result<Vec<CreateAclOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn delete_acls<'life0, 'life1, 'async_trait>( &'life0 mut self, filters: &'life1 [AclEntryFilter], ) -> Pin<Box<dyn Future<Output = Result<Vec<DeleteAclFilterOutcome>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn describe_user_quotas<'life0, 'life1, 'async_trait>( &'life0 mut self, username: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<UserQuotaConfig, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn alter_user_quotas<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, username: &'life1 str, ops: &'life2 [QuotaOp], validate_only: bool, ) -> Pin<Box<dyn Future<Output = Result<Option<KafkaError>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn create_delegation_token_as_owner<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, owner_principal_name: &'life1 str, renewers: &'life2 [String], max_lifetime_ms: i64, ) -> Pin<Box<dyn Future<Output = Result<DelegationToken, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn renew_delegation_token<'life0, 'life1, 'async_trait>( &'life0 mut self, hmac: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = Result<DelegationToken, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn expire_delegation_token<'life0, 'life1, 'async_trait>( &'life0 mut self, hmac: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = Result<(), AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn describe_delegation_tokens_owned_by<'life0, 'life1, 'async_trait>( &'life0 mut self, owner_principal: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<DelegationToken>, AdminError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,