Trait Admin 

pub trait Admin: Send + Sync {
    // Required methods
    fn create_topics_with_options(
        &self,
        new_topics: &[NewTopic],
        options: CreateTopicsOptions,
    ) -> CreateTopicsResult;
    fn delete_topics_with_topic_collection_options(
        &self,
        topics: TopicCollection,
        options: DeleteTopicsOptions,
    ) -> DeleteTopicsResult;
    fn list_topics_with_options(
        &self,
        options: ListTopicsOptions,
    ) -> ListTopicsResult;
    fn describe_topics_with_topics_options(
        &self,
        topics: TopicCollection,
        options: DescribeTopicsOptions,
    ) -> DescribeTopicsResult;
    fn create_partitions_with_options(
        &self,
        new_partitions: &HashMap<String, NewPartitions>,
        options: CreatePartitionsOptions,
    ) -> CreatePartitionsResult;
    fn delete_records_with_options(
        &self,
        records_to_delete: &HashMap<TopicPartition, RecordsToDelete>,
        options: DeleteRecordsOptions,
    ) -> DeleteRecordsResult;
    fn unregister_controller_with_options(
        &self,
        controller_id: i32,
        options: UnregisterControllerOptions,
    ) -> UnregisterControllerResult;
    fn describe_producers_with_options(
        &self,
        partitions: &[TopicPartition],
        options: DescribeProducersOptions,
    ) -> DescribeProducersResult;
    fn abort_transaction_with_options(
        &self,
        spec: AbortTransactionSpec,
        options: AbortTransactionOptions,
    ) -> AbortTransactionResult;
    fn describe_transactions_with_options(
        &self,
        transactional_ids: &[String],
        options: DescribeTransactionsOptions,
    ) -> DescribeTransactionsResult;
    fn fence_producers_with_options(
        &self,
        transactional_ids: &[String],
        options: FenceProducersOptions,
    ) -> FenceProducersResult;
    fn list_transactions_with_options(
        &self,
        options: ListTransactionsOptions,
    ) -> ListTransactionsResult;
    fn force_terminate_transaction_with_options(
        &self,
        transactional_id: &str,
        options: TerminateTransactionOptions,
    ) -> TerminateTransactionResult;
    fn describe_cluster_with_options(
        &self,
        options: DescribeClusterOptions,
    ) -> DescribeClusterResult;
    fn describe_configs_with_options(
        &self,
        config_resources: &[ConfigResource],
        options: DescribeConfigsOptions,
    ) -> DescribeConfigsResult;
    fn incremental_alter_configs_with_options(
        &self,
        configs: &HashMap<ConfigResource, Vec<AlterConfigOp>>,
        options: AlterConfigsOptions,
    ) -> AlterConfigsResult;
    fn list_config_resources_with_options(
        &self,
        config_resource_types: &HashSet<Type>,
        options: ListConfigResourcesOptions,
    ) -> ListConfigResourcesResult;
    fn describe_log_dirs_with_options(
        &self,
        brokers: &[i32],
        options: DescribeLogDirsOptions,
    ) -> DescribeLogDirsResult;
    fn alter_replica_log_dirs_with_options(
        &self,
        replica_assignment: &HashMap<TopicPartitionReplica, String>,
        options: AlterReplicaLogDirsOptions,
    ) -> AlterReplicaLogDirsResult;
    fn describe_replica_log_dirs_with_options(
        &self,
        replicas: &[TopicPartitionReplica],
        options: DescribeReplicaLogDirsOptions,
    ) -> DescribeReplicaLogDirsResult;
    fn elect_leaders_with_options(
        &self,
        election_type: ElectionType,
        partitions: Option<HashSet<TopicPartition>>,
        options: ElectLeadersOptions,
    ) -> ElectLeadersResult;
    fn alter_partition_reassignments_with_options(
        &self,
        reassignments: &HashMap<TopicPartition, Option<NewPartitionReassignment>>,
        options: AlterPartitionReassignmentsOptions,
    ) -> AlterPartitionReassignmentsResult;
    fn list_partition_reassignments_with_partitions_options(
        &self,
        partitions: Option<HashSet<TopicPartition>>,
        options: ListPartitionReassignmentsOptions,
    ) -> ListPartitionReassignmentsResult;
    fn list_offsets_with_options(
        &self,
        topic_partition_offsets: &HashMap<TopicPartition, OffsetSpec>,
        options: ListOffsetsOptions,
    ) -> ListOffsetsResult;
    fn list_groups_with_options(
        &self,
        options: ListGroupsOptions,
    ) -> ListGroupsResult;
    fn describe_consumer_groups_with_options(
        &self,
        group_ids: &[String],
        options: DescribeConsumerGroupsOptions,
    ) -> DescribeConsumerGroupsResult;
    fn describe_classic_groups_with_options(
        &self,
        group_ids: &[String],
        options: DescribeClassicGroupsOptions,
    ) -> DescribeClassicGroupsResult;
    fn list_consumer_group_offsets_with_group_specs_options(
        &self,
        group_specs: &HashMap<String, ListConsumerGroupOffsetsSpec>,
        options: ListConsumerGroupOffsetsOptions,
    ) -> ListConsumerGroupOffsetsResult;
    fn alter_consumer_group_offsets_with_options(
        &self,
        group_id: &str,
        offsets: &HashMap<TopicPartition, OffsetAndMetadata>,
        options: AlterConsumerGroupOffsetsOptions,
    ) -> AlterConsumerGroupOffsetsResult;
    fn delete_consumer_group_offsets_with_options(
        &self,
        group_id: &str,
        partitions: &HashSet<TopicPartition>,
        options: DeleteConsumerGroupOffsetsOptions,
    ) -> DeleteConsumerGroupOffsetsResult;
    fn delete_consumer_groups_with_options(
        &self,
        group_ids: &[String],
        options: DeleteConsumerGroupsOptions,
    ) -> DeleteConsumerGroupsResult;
    fn remove_members_from_consumer_group_with_options(
        &self,
        group_id: &str,
        options: RemoveMembersFromConsumerGroupOptions,
    ) -> RemoveMembersFromConsumerGroupResult;
    fn create_acls_with_options(
        &self,
        acls: &[AclBinding],
        options: CreateAclsOptions,
    ) -> CreateAclsResult;
    fn describe_acls_with_options(
        &self,
        filter: &AclBindingFilter,
        options: DescribeAclsOptions,
    ) -> DescribeAclsResult;
    fn delete_acls_with_options(
        &self,
        filters: &[AclBindingFilter],
        options: DeleteAclsOptions,
    ) -> DeleteAclsResult;
    fn describe_client_quotas_with_options(
        &self,
        filter: &ClientQuotaFilter,
        options: DescribeClientQuotasOptions,
    ) -> DescribeClientQuotasResult;
    fn alter_client_quotas_with_options(
        &self,
        entries: &[ClientQuotaAlteration],
        options: AlterClientQuotasOptions,
    ) -> AlterClientQuotasResult;
    fn describe_user_scram_credentials_with_users_options(
        &self,
        users: &[String],
        options: DescribeUserScramCredentialsOptions,
    ) -> DescribeUserScramCredentialsResult;
    fn alter_user_scram_credentials_with_options(
        &self,
        alterations: &[UserScramCredentialAlteration],
        options: AlterUserScramCredentialsOptions,
    ) -> AlterUserScramCredentialsResult;
    fn create_delegation_token_with_options(
        &self,
        options: CreateDelegationTokenOptions,
    ) -> CreateDelegationTokenResult;
    fn renew_delegation_token_with_options(
        &self,
        hmac: &[u8],
        options: RenewDelegationTokenOptions,
    ) -> RenewDelegationTokenResult;
    fn expire_delegation_token_with_options(
        &self,
        hmac: &[u8],
        options: ExpireDelegationTokenOptions,
    ) -> ExpireDelegationTokenResult;
    fn describe_delegation_token_with_options(
        &self,
        options: DescribeDelegationTokenOptions,
    ) -> DescribeDelegationTokenResult;
    fn describe_features_with_options(
        &self,
        options: DescribeFeaturesOptions,
    ) -> DescribeFeaturesResult;
    fn update_features_with_options(
        &self,
        feature_updates: &HashMap<String, FeatureUpdate>,
        options: UpdateFeaturesOptions,
    ) -> Result<UpdateFeaturesResult, Error>;
    fn close_with_timeout<'life0, 'async_trait>(
        &'life0 self,
        timeout: Duration,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn create_topics(&self, new_topics: &[NewTopic]) -> CreateTopicsResult { ... }
    fn delete_topics_with_string_slice(
        &self,
        topics: &[String],
    ) -> DeleteTopicsResult { ... }
    fn delete_topics_with_string_slice_options(
        &self,
        topics: &[String],
        options: DeleteTopicsOptions,
    ) -> DeleteTopicsResult { ... }
    fn delete_topics_with_topic_collection(
        &self,
        topics: TopicCollection,
    ) -> DeleteTopicsResult { ... }
    fn list_topics(&self) -> ListTopicsResult { ... }
    fn describe_topics_with_topic_names(
        &self,
        topic_names: &[String],
    ) -> DescribeTopicsResult { ... }
    fn describe_topics_with_topic_names_options(
        &self,
        topic_names: &[String],
        options: DescribeTopicsOptions,
    ) -> DescribeTopicsResult { ... }
    fn describe_topics_with_topics(
        &self,
        topics: TopicCollection,
    ) -> DescribeTopicsResult { ... }
    fn create_partitions(
        &self,
        new_partitions: &HashMap<String, NewPartitions>,
    ) -> CreatePartitionsResult { ... }
    fn delete_records(
        &self,
        records_to_delete: &HashMap<TopicPartition, RecordsToDelete>,
    ) -> DeleteRecordsResult { ... }
    fn unregister_controller(
        &self,
        controller_id: i32,
    ) -> UnregisterControllerResult { ... }
    fn describe_producers(
        &self,
        partitions: &[TopicPartition],
    ) -> DescribeProducersResult { ... }
    fn abort_transaction(
        &self,
        spec: AbortTransactionSpec,
    ) -> AbortTransactionResult { ... }
    fn describe_transactions(
        &self,
        transactional_ids: &[String],
    ) -> DescribeTransactionsResult { ... }
    fn fence_producers(
        &self,
        transactional_ids: &[String],
    ) -> FenceProducersResult { ... }
    fn list_transactions(&self) -> ListTransactionsResult { ... }
    fn force_terminate_transaction(
        &self,
        transactional_id: &str,
    ) -> TerminateTransactionResult { ... }
    fn describe_cluster(&self) -> DescribeClusterResult { ... }
    fn describe_configs(
        &self,
        config_resources: &[ConfigResource],
    ) -> DescribeConfigsResult { ... }
    fn incremental_alter_configs(
        &self,
        configs: &HashMap<ConfigResource, Vec<AlterConfigOp>>,
    ) -> AlterConfigsResult { ... }
    fn list_config_resources(&self) -> ListConfigResourcesResult { ... }
    fn describe_log_dirs(&self, brokers: &[i32]) -> DescribeLogDirsResult { ... }
    fn alter_replica_log_dirs(
        &self,
        replica_assignment: &HashMap<TopicPartitionReplica, String>,
    ) -> AlterReplicaLogDirsResult { ... }
    fn describe_replica_log_dirs(
        &self,
        replicas: &[TopicPartitionReplica],
    ) -> DescribeReplicaLogDirsResult { ... }
    fn elect_leaders(
        &self,
        election_type: ElectionType,
        partitions: Option<HashSet<TopicPartition>>,
    ) -> ElectLeadersResult { ... }
    fn alter_partition_reassignments(
        &self,
        reassignments: &HashMap<TopicPartition, Option<NewPartitionReassignment>>,
    ) -> AlterPartitionReassignmentsResult { ... }
    fn list_partition_reassignments(&self) -> ListPartitionReassignmentsResult { ... }
    fn list_partition_reassignments_with_partitions(
        &self,
        partitions: HashSet<TopicPartition>,
    ) -> ListPartitionReassignmentsResult { ... }
    fn list_partition_reassignments_with_options(
        &self,
        options: ListPartitionReassignmentsOptions,
    ) -> ListPartitionReassignmentsResult { ... }
    fn list_offsets(
        &self,
        topic_partition_offsets: &HashMap<TopicPartition, OffsetSpec>,
    ) -> ListOffsetsResult { ... }
    fn list_groups(&self) -> ListGroupsResult { ... }
    fn describe_consumer_groups(
        &self,
        group_ids: &[String],
    ) -> DescribeConsumerGroupsResult { ... }
    fn describe_classic_groups(
        &self,
        group_ids: &[String],
    ) -> DescribeClassicGroupsResult { ... }
    fn list_consumer_group_offsets_with_group_id(
        &self,
        group_id: &str,
    ) -> ListConsumerGroupOffsetsResult { ... }
    fn list_consumer_group_offsets_with_group_id_options(
        &self,
        group_id: &str,
        options: ListConsumerGroupOffsetsOptions,
    ) -> ListConsumerGroupOffsetsResult { ... }
    fn list_consumer_group_offsets_with_group_specs(
        &self,
        group_specs: &HashMap<String, ListConsumerGroupOffsetsSpec>,
    ) -> ListConsumerGroupOffsetsResult { ... }
    fn alter_consumer_group_offsets(
        &self,
        group_id: &str,
        offsets: &HashMap<TopicPartition, OffsetAndMetadata>,
    ) -> AlterConsumerGroupOffsetsResult { ... }
    fn delete_consumer_group_offsets(
        &self,
        group_id: &str,
        partitions: &HashSet<TopicPartition>,
    ) -> DeleteConsumerGroupOffsetsResult { ... }
    fn delete_consumer_groups(
        &self,
        group_ids: &[String],
    ) -> DeleteConsumerGroupsResult { ... }
    fn create_acls(&self, acls: &[AclBinding]) -> CreateAclsResult { ... }
    fn describe_acls(&self, filter: &AclBindingFilter) -> DescribeAclsResult { ... }
    fn delete_acls(&self, filters: &[AclBindingFilter]) -> DeleteAclsResult { ... }
    fn describe_client_quotas(
        &self,
        filter: &ClientQuotaFilter,
    ) -> DescribeClientQuotasResult { ... }
    fn alter_client_quotas(
        &self,
        entries: &[ClientQuotaAlteration],
    ) -> AlterClientQuotasResult { ... }
    fn describe_user_scram_credentials(
        &self,
    ) -> DescribeUserScramCredentialsResult { ... }
    fn describe_user_scram_credentials_with_users(
        &self,
        users: &[String],
    ) -> DescribeUserScramCredentialsResult { ... }
    fn alter_user_scram_credentials(
        &self,
        alterations: &[UserScramCredentialAlteration],
    ) -> AlterUserScramCredentialsResult { ... }
    fn create_delegation_token(&self) -> CreateDelegationTokenResult { ... }
    fn renew_delegation_token(&self, hmac: &[u8]) -> RenewDelegationTokenResult { ... }
    fn expire_delegation_token(
        &self,
        hmac: &[u8],
    ) -> ExpireDelegationTokenResult { ... }
    fn describe_delegation_token(&self) -> DescribeDelegationTokenResult { ... }
    fn describe_features(&self) -> DescribeFeaturesResult { ... }
    fn update_features(
        &self,
        feature_updates: &HashMap<String, FeatureUpdate>,
    ) -> Result<UpdateFeaturesResult, Error> { ... }
    fn close<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
}

The administrative client for Kafka, which supports managing and inspecting topics, brokers, configurations and ACLs.

Instances returned from AdminClient::create are guaranteed to be thread safe. However, the KafkaFutures returned from request methods are completed by a single task so it is important that any code which executes on that task when they complete (using KafkaFuture::then_apply, for example) doesn’t block for too long. If necessary, processing of results should be passed to another task.

The operations exposed by Admin follow a consistent pattern:

• Admin instances should be created using AdminClient::create.
• Each operation typically has two overloaded methods, one which uses a default set of options and an overloaded method where the last parameter is an explicit options object.
• The operation method’s first parameter is a collection of items to perform the operation on. Batching multiple requests into a single call is more efficient and should be preferred over multiple calls to the same method.
• The operation methods execute asynchronously.
• Each xxx operation method returns an XxxResult class with methods which expose KafkaFuture for accessing the result(s) of the operation.
• Typically an all() method is provided for getting the overall success/failure of the batch and a values() method provided access to each item in a request batch. Other methods may also be provided.
• For synchronous behavior, await KafkaFuture::get.

Here is a simple example of using an Admin client instance to create a new topic:

use std::collections::{BTreeMap, HashMap};

use confluent_kafka::admin::{AdminClient, AdminClientConfig, NewTopic};

let props = HashMap::from([(
    AdminClientConfig::BOOTSTRAP_SERVERS_CONFIG.to_string(),
    "localhost:9092".to_string(),
)]);
let admin = AdminClient::create(AdminClientConfig::new(&props)?)?;
let topic_name = "my-topic";
// Create a compacted topic
let result = admin.create_topics(&[NewTopic::with_num_partitions_replication_factor(
    topic_name,
    Some(12),
    Some(3),
)
.set_configs(Some(BTreeMap::from([("cleanup.policy".to_string(), "compact".to_string())])))]);

// Call values() to get the result for a specific topic
let future = result.values()[topic_name].clone();

// Await get() until the topic creation is complete or has failed
future.get().await?;
admin.close().await;

Bootstrap and balancing

The bootstrap.servers config passed to AdminClient::create is only used for discovering the brokers in the cluster, which the client will then connect to as needed. As such, it is sufficient to include only two or three broker addresses to cope with the possibility of brokers being unavailable.

Different operations necessitate requests being sent to different nodes in the cluster. For example Self::create_topics communicates with the controller, but Self::describe_topics_with_topic_names can talk to any broker. When the recipient does not matter the instance will try to use the broker with the fewest outstanding requests.

The client will transparently retry certain errors which are usually transient. For example if the request for create_topics() get sent to a node which was not the controller the metadata would be refreshed and the request re-sent to the controller.

Broker Compatibility

The minimum broker version required is 0.10.0.0. Methods with stricter requirements will specify the minimum broker version required.

Translation notes

The request methods are plain fns that return at once; only close and close_with_timeout are async, as Java’s close joins the admin thread.

Required Methods

fn create_topics_with_options( &self, new_topics: &[NewTopic], options: CreateTopicsOptions, ) -> CreateTopicsResult

Create a batch of new topics.

This operation is not transactional so it may succeed for some topics while fail for others.

It may take several seconds after CreateTopicsResult returns success for all the brokers to become aware that the topics have been created. During this time, Self::list_topics and Self::describe_topics_with_topic_names may not return information about the new topics.

This operation is supported by brokers with version 0.10.1.0 or higher. The validateOnly option is supported from version 0.10.2.0.

Arguments

• new_topics - The new topics to create.
• options - The options to use when creating the new topics.

Returns

The CreateTopicsResult.

fn delete_topics_with_topic_collection_options( &self, topics: TopicCollection, options: DeleteTopicsOptions, ) -> DeleteTopicsResult

Delete a batch of topics.

This operation is not transactional so it may succeed for some topics while fail for others.

It may take several seconds after the DeleteTopicsResult returns success for all the brokers to become aware that the topics are gone. During this time, Self::list_topics and Self::describe_topics_with_topic_names may continue to return information about the deleted topics.

If delete.topic.enable is set to false on the brokers, an error will be returned to the client indicating that topic deletion is disabled.

When using topic IDs, this operation is supported by brokers with inter-broker protocol 2.8 or higher. When using topic names, this operation is supported by brokers with version 0.10.1.0 or higher.

Arguments

• topics - The topics to delete.
• options - The options to use when deleting the topics.

Returns

The DeleteTopicsResult.

fn list_topics_with_options( &self, options: ListTopicsOptions, ) -> ListTopicsResult

List the topics available in the cluster.

Arguments

• options - The options to use when listing the topics.

Returns

The ListTopicsResult.

fn describe_topics_with_topics_options( &self, topics: TopicCollection, options: DescribeTopicsOptions, ) -> DescribeTopicsResult

Describe some topics in the cluster. When using topic IDs, this operation is supported by brokers with version 3.1.0 or higher.

Arguments

• topics - The topics to describe.
• options - The options to use when describing the topics.

Returns

The DescribeTopicsResult.

fn create_partitions_with_options( &self, new_partitions: &HashMap<String, NewPartitions>, options: CreatePartitionsOptions, ) -> CreatePartitionsResult

Increase the number of partitions of the topics given as the keys of new_partitions according to the corresponding values. If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected.

This operation is not transactional so it may succeed for some topics while fail for others.

It may take several seconds after this method returns success for all the brokers to become aware that the partitions have been created. During this time, Self::describe_topics_with_topic_names may not return information about the new partitions.

This operation is supported by brokers with version 1.0.0 or higher.

The following errors can be anticipated when calling get() on the futures obtained from the values() method of the returned CreatePartitionsResult

• Error::is_authorization_error if the authenticated user is not authorized to alter the topic
• Error::is_timeout_error if the request was not completed in within the given CreatePartitionsOptions::timeout_ms.
• Error::is_reassignment_in_progress_error if a partition reassignment is currently in progress
• Error::is_broker_not_available_error if the requested NewPartitions::assignments contain a broker that is currently unavailable.
• Error::is_invalid_replication_factor_error if no NewPartitions::assignments are given and it is impossible for the broker to assign replicas with the topics replication factor.
• Subclasses of Error::is_kafka_error if the request is invalid in some way.
• Error::is_invalid_partitions_error if the requested partition count is less than or equal to the current partition count.

Arguments

• new_partitions - The topics which should have new partitions created, and corresponding parameters for the created partitions.
• options - The options to use when creating the new partitions.

Returns

The CreatePartitionsResult.

fn delete_records_with_options( &self, records_to_delete: &HashMap<TopicPartition, RecordsToDelete>, options: DeleteRecordsOptions, ) -> DeleteRecordsResult

Delete records whose offset is smaller than the given offset of the corresponding partition.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• records_to_delete - The topic partitions and related offsets from which records deletion starts.
• options - The options to use when deleting records.

Returns

The DeleteRecordsResult.

fn unregister_controller_with_options( &self, controller_id: i32, options: UnregisterControllerOptions, ) -> UnregisterControllerResult

Unregister a controller. The following errors can be anticipated when calling get() on the future from the returned UnregisterControllerResult:

• Error::is_timeout_error If the request timed out before the unregister operation could finish.
• Error::is_unsupported_version_error If the software is too old to support the unregistration API.
• Error::is_controller_id_not_registered_error If the requested controller id is not currently registered.
• Error::is_not_controller_error If the request does not arrive at the active controller.
• Error::is_invalid_request_error If the request tries to unregister the current active controller id.

Arguments

• controller_id - the controller id to unregister.
• options - the options to use.

Returns

the UnregisterControllerResult containing the result

fn describe_producers_with_options( &self, partitions: &[TopicPartition], options: DescribeProducersOptions, ) -> DescribeProducersResult

Describe active producer state on a set of topic partitions. Unless a specific broker is requested through DescribeProducersOptions::broker_id, this will query the partition leader to find the producer state.

Arguments

• partitions - The set of partitions to query
• options - Options to control the method behavior

Returns

The result

fn abort_transaction_with_options( &self, spec: AbortTransactionSpec, options: AbortTransactionOptions, ) -> AbortTransactionResult

Forcefully abort a transaction which is open on a topic partition. This will send a WriteTxnMarkers request to the partition leader in order to abort the transaction. This requires administrative privileges.

Arguments

• spec - The transaction specification including topic partition and producer details
• options - Options to control the method behavior (including filters)

Returns

The result

fn describe_transactions_with_options( &self, transactional_ids: &[String], options: DescribeTransactionsOptions, ) -> DescribeTransactionsResult

Describe the state of a set of transactional IDs from the respective transaction coordinators, which are dynamically discovered.

Arguments

• transactional_ids - The set of transactional IDs to query
• options - Options to control the method behavior

Returns

The result

fn fence_producers_with_options( &self, transactional_ids: &[String], options: FenceProducersOptions, ) -> FenceProducersResult

Fence out all active producers that use any of the provided transactional IDs.

Arguments

• transactional_ids - The IDs of the producers to fence.
• options - The options to use when fencing the producers.

Returns

The FenceProducersResult.

fn list_transactions_with_options( &self, options: ListTransactionsOptions, ) -> ListTransactionsResult

List active transactions in the cluster. This will query all potential transaction coordinators in the cluster and collect the state of all transactions. Users should typically attempt to reduce the size of the result set using ListTransactionsOptions::filter_producer_ids or ListTransactionsOptions::filter_states or ListTransactionsOptions::filter_on_duration.

Arguments

• options - Options to control the method behavior (including filters)

Returns

The result

fn force_terminate_transaction_with_options( &self, transactional_id: &str, options: TerminateTransactionOptions, ) -> TerminateTransactionResult

Force terminate a transaction for the given transactional ID. This operation aborts any ongoing transaction associated with the transactional ID. It’s similar to fenceProducers but only targets a single transactional ID to handle long-running transactions when 2PC is enabled.

Arguments

• transactional_id - The ID of the transaction to terminate.
• options - The options to use when terminating the transaction.

Returns

The TerminateTransactionResult.

fn describe_cluster_with_options( &self, options: DescribeClusterOptions, ) -> DescribeClusterResult

Get information about the nodes in the cluster.

To obtain broker cluster information, you must configure AdminClientConfig::BOOTSTRAP_SERVERS_CONFIG. To obtain controller cluster information, you must configure AdminClientConfig::BOOTSTRAP_CONTROLLERS_CONFIG.

Arguments

• options - The options to use when getting information about the cluster.

Returns

The DescribeClusterResult.

fn describe_configs_with_options( &self, config_resources: &[ConfigResource], options: DescribeConfigsOptions, ) -> DescribeConfigsResult

Get the configuration for the specified resources.

The returned configuration includes default values and the ConfigEntry::is_default method can be used to distinguish them from user supplied values.

The value of config entries where ConfigEntry::is_sensitive is true is always None so that sensitive information is not disclosed.

Config entries where ConfigEntry::is_read_only is true cannot be updated.

The different behavior of nonexistent resource:

• config_resource::Type::Broker: fails with Error::is_timeout_error
• config_resource::Type::Topic: fails with Error::is_unknown_topic_or_partition_error
• config_resource::Type::Group: just return default configs even if the target group is nonexistent
• config_resource::Type::BrokerLogger: fails with Error::is_timeout_error
• config_resource::Type::ClientMetrics: will return empty configs

Note that you cannot describe broker configs or broker logger using AdminClientConfig::BOOTSTRAP_CONTROLLERS_CONFIG, and you cannot describe controller configs or controller logger using AdminClientConfig::BOOTSTRAP_SERVERS_CONFIG.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• config_resources - The resources (topic and broker resource types are currently supported)
• options - The options to use when describing configs

Returns

The DescribeConfigsResult

fn incremental_alter_configs_with_options( &self, configs: &HashMap<ConfigResource, Vec<AlterConfigOp>>, options: AlterConfigsOptions, ) -> AlterConfigsResult

Incrementally update the configuration for the specified resources.

Updates are not transactional so they may succeed for some resources while fail for others. The configs for a particular resource are updated atomically.

The following errors can be anticipated when calling get() on the futures obtained from the returned AlterConfigsResult:

• Error::is_cluster_authorization_error if the authenticated user didn’t have alter access to the cluster.
• Error::is_topic_authorization_error if the authenticated user didn’t have alter access to the Topic.
• Error::is_unknown_topic_or_partition_error if the Topic doesn’t exist.
• Error::is_invalid_request_error if the request details are invalid. e.g., a configuration key was specified more than once for a resource

This operation is supported by brokers with version 2.3.0 or higher.

Arguments

• configs - The resources with their configs
• options - The options to use when altering configs

Returns

The AlterConfigsResult

fn list_config_resources_with_options( &self, config_resource_types: &HashSet<Type>, options: ListConfigResourcesOptions, ) -> ListConfigResourcesResult

List the configuration resources available in the cluster which matches config resource type. If no config resource types are specified, all configuration resources will be listed.

Arguments

• config_resource_types - The set of configuration resource types to list.
• options - The options to use when listing the configuration resources.

Returns

The ListConfigurationResourcesResult.

fn describe_log_dirs_with_options( &self, brokers: &[i32], options: DescribeLogDirsOptions, ) -> DescribeLogDirsResult

Query the information of all log directories on the given set of brokers

This operation is supported by brokers with version 1.0.0 or higher.

Arguments

• brokers - A list of brokers
• options - The options to use when querying log dir info

Returns

The DescribeLogDirsResult

fn alter_replica_log_dirs_with_options( &self, replica_assignment: &HashMap<TopicPartitionReplica, String>, options: AlterReplicaLogDirsOptions, ) -> AlterReplicaLogDirsResult

Change the log directory for the specified replicas. If the replica does not exist on the broker, the result shows REPLICA_NOT_AVAILABLE for the given replica and the replica will be created in the given log directory on the broker when it is created later. If the replica already exists on the broker, the replica will be moved to the given log directory if it is not already there. For detailed result, inspect the returned AlterReplicaLogDirsResult instance.

This operation is not transactional so it may succeed for some replicas while fail for others.

This operation is supported by brokers with version 1.1.0 or higher.

Arguments

• replica_assignment - The replicas with their log directory absolute path
• options - The options to use when changing replica dir

Returns

The AlterReplicaLogDirsResult

fn describe_replica_log_dirs_with_options( &self, replicas: &[TopicPartitionReplica], options: DescribeReplicaLogDirsOptions, ) -> DescribeReplicaLogDirsResult

Query the replica log directory information for the specified replicas.

This operation is supported by brokers with version 1.0.0 or higher.

Arguments

• replicas - The replicas to query
• options - The options to use when querying replica log dir info

Returns

The DescribeReplicaLogDirsResult

fn elect_leaders_with_options( &self, election_type: ElectionType, partitions: Option<HashSet<TopicPartition>>, options: ElectLeadersOptions, ) -> ElectLeadersResult

Elect a replica as leader for the given partitions, or for all partitions if the argument to partitions is None.

This operation is not transactional so it may succeed for some partitions while fail for others.

It may take several seconds after this method returns success for all the brokers in the cluster to become aware that the partitions have new leaders. During this time, Self::describe_topics_with_topic_names may not return information about the partitions’ new leaders.

This operation is supported by brokers with version 2.2.0 or later if preferred election is use; otherwise the brokers most be 2.4.0 or higher.

The following errors can be anticipated when calling get() on the future obtained from the returned ElectLeadersResult:

• Error::is_cluster_authorization_error if the authenticated user didn’t have alter access to the cluster.
• Error::is_unknown_topic_or_partition_error if the topic or partition did not exist within the cluster.
• Error::is_invalid_topic_error if the topic was already queued for deletion.
• Error::is_not_controller_error if the request was sent to a broker that was not the controller for the cluster.
• Error::is_timeout_error if the request timed out before the election was complete.
• Error::is_leader_not_available_error if the preferred leader was not alive or not in the ISR.

Arguments

• election_type - The type of election to conduct.
• partitions - The topics and partitions for which to conduct elections.
• options - The options to use when electing the leaders.

Returns

The ElectLeadersResult.

fn alter_partition_reassignments_with_options( &self, reassignments: &HashMap<TopicPartition, Option<NewPartitionReassignment>>, options: AlterPartitionReassignmentsOptions, ) -> AlterPartitionReassignmentsResult

Change the reassignments for one or more partitions. Providing None will revert the reassignment for the associated partition.

The following errors can be anticipated when calling get() on the futures obtained from the returned AlterPartitionReassignmentsResult:

• Error::is_cluster_authorization_error If the authenticated user didn’t have alter access to the cluster.
• Error::is_unknown_topic_or_partition_error If the topic or partition does not exist within the cluster.
• Error::is_timeout_error if the request timed out before the controller could record the new assignments.
• Error::is_invalid_replica_assignment_error If the specified assignment was not valid.
• Error::is_invalid_replication_factor_error If the replication factor was changed in an invalid way. Only thrown when AlterPartitionReassignmentsOptions::allow_replication_factor_change is set to false and the request is attempting to alter reassignments (not cancel)
• Error::is_unsupported_version_error If AlterPartitionReassignmentsOptions::allow_replication_factor_change was changed outside the default and the server does not support the option (e.g due to an old Kafka version).
• Error::is_no_reassignment_in_progress_error If there was an attempt to cancel a reassignment for a partition which was not being reassigned.

Arguments

• reassignments - The reassignments to add, modify, or remove. See NewPartitionReassignment.
• options - The options to use.

Returns

The result.

fn list_partition_reassignments_with_partitions_options( &self, partitions: Option<HashSet<TopicPartition>>, options: ListPartitionReassignmentsOptions, ) -> ListPartitionReassignmentsResult

List the current reassignments for the given partitions.

The following errors can be anticipated when calling get() on the futures obtained from the returned ListPartitionReassignmentsResult:

• Error::is_cluster_authorization_error If the authenticated user doesn’t have alter access to the cluster.
• Error::is_unknown_topic_or_partition_error If a given topic or partition does not exist.
• Error::is_timeout_error If the request timed out before the controller could list the current reassignments.

Arguments

• partitions - the partitions we want to get reassignment for, or None if we want to get the reassignments for all partitions in the cluster
• options - The options to use.

Returns

The result.

fn list_offsets_with_options( &self, topic_partition_offsets: &HashMap<TopicPartition, OffsetSpec>, options: ListOffsetsOptions, ) -> ListOffsetsResult

List offset for the specified partitions. This operation enables to find the beginning offset, end offset as well as the offset matching a timestamp in partitions.

Arguments

• topic_partition_offsets - The mapping from partition to the OffsetSpec to look up.
• options - The options to use when retrieving the offsets

Returns

The ListOffsetsResult.

fn list_groups_with_options( &self, options: ListGroupsOptions, ) -> ListGroupsResult

List the groups available in the cluster.

Arguments

• options - The options to use when listing the groups.

Returns

The ListGroupsResult.

fn describe_consumer_groups_with_options( &self, group_ids: &[String], options: DescribeConsumerGroupsOptions, ) -> DescribeConsumerGroupsResult

Describe some consumer groups in the cluster.

Arguments

• group_ids - The IDs of the groups to describe.
• options - The options to use when describing the groups.

Returns

The DescribeConsumerGroupsResult.

fn describe_classic_groups_with_options( &self, group_ids: &[String], options: DescribeClassicGroupsOptions, ) -> DescribeClassicGroupsResult

Describe some classic groups in the cluster.

Arguments

• group_ids - The IDs of the groups to describe.
• options - The options to use when describing the groups.

Returns

The DescribeClassicGroupsResult.

fn list_consumer_group_offsets_with_group_specs_options( &self, group_specs: &HashMap<String, ListConsumerGroupOffsetsSpec>, options: ListConsumerGroupOffsetsOptions, ) -> ListConsumerGroupOffsetsResult

List the consumer group offsets available in the cluster for the specified consumer groups.

Arguments

• group_specs - Map of consumer group ids to a spec that specifies the topic partitions of the group to list offsets for.
• options - The options to use when listing the consumer group offsets.

Returns

The ListConsumerGroupOffsetsResult

fn alter_consumer_group_offsets_with_options( &self, group_id: &str, offsets: &HashMap<TopicPartition, OffsetAndMetadata>, options: AlterConsumerGroupOffsetsOptions, ) -> AlterConsumerGroupOffsetsResult

Alters offsets for the specified group. In order to succeed, the group must be empty.

This operation is not transactional so it may succeed for some partitions while fail for others.

Arguments

• group_id - The group for which to alter offsets.
• offsets - A map of offsets by partition with associated metadata. Partitions not specified in the map are ignored.
• options - The options to use when altering the offsets.

Returns

The AlterOffsetsResult.

fn delete_consumer_group_offsets_with_options( &self, group_id: &str, partitions: &HashSet<TopicPartition>, options: DeleteConsumerGroupOffsetsOptions, ) -> DeleteConsumerGroupOffsetsResult

Delete committed offsets for a set of partitions in a consumer group. This will succeed at the partition level only if the group is not actively subscribed to the corresponding topic.

Arguments

• group_id - The group to delete the offsets of.
• partitions - The partitions whose committed offsets to delete.
• options - The options to use when deleting offsets in a consumer group.

Returns

The DeleteConsumerGroupOffsetsResult.

fn delete_consumer_groups_with_options( &self, group_ids: &[String], options: DeleteConsumerGroupsOptions, ) -> DeleteConsumerGroupsResult

Delete consumer groups from the cluster.

Arguments

• group_ids - The ids of the groups to delete.
• options - The options to use when deleting a consumer group.

Returns

The DeleteConsumerGroupsResult.

fn remove_members_from_consumer_group_with_options( &self, group_id: &str, options: RemoveMembersFromConsumerGroupOptions, ) -> RemoveMembersFromConsumerGroupResult

Remove members from the consumer group by given member identities.

For possible error codes, refer to LeaveGroupResponse.

Arguments

• group_id - The ID of the group to remove member from.
• options - The options to carry removing members’ information.

Returns

The MembershipChangeResult.

fn create_acls_with_options( &self, acls: &[AclBinding], options: CreateAclsOptions, ) -> CreateAclsResult

Creates access control lists (ACLs) which are bound to specific resources.

This operation is not transactional so it may succeed for some ACLs while fail for others.

If you attempt to add an ACL that duplicates an existing ACL, no error will be raised, but no changes will be made.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• acls - The ACLs to create
• options - The options to use when creating the ACLs.

Returns

The CreateAclsResult.

fn describe_acls_with_options( &self, filter: &AclBindingFilter, options: DescribeAclsOptions, ) -> DescribeAclsResult

Lists access control lists (ACLs) according to the supplied filter.

Note: it may take some time for changes made by createAcls or deleteAcls to be reflected in the output of describeAcls.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• filter - The filter to use.
• options - The options to use when listing the ACLs.

Returns

The DescribeAclsResult.

fn delete_acls_with_options( &self, filters: &[AclBindingFilter], options: DeleteAclsOptions, ) -> DeleteAclsResult

Deletes access control lists (ACLs) according to the supplied filters.

This operation is not transactional so it may succeed for some ACLs while fail for others.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• filters - The filters to use.
• options - The options to use when deleting the ACLs.

Returns

The DeleteAclsResult.

fn describe_client_quotas_with_options( &self, filter: &ClientQuotaFilter, options: DescribeClientQuotasOptions, ) -> DescribeClientQuotasResult

Describes all entities matching the provided filter that have at least one client quota configuration value defined.

The following errors can be anticipated when calling get() on the future from the returned DescribeClientQuotasResult:

• Error::is_cluster_authorization_error If the authenticated user didn’t have describe access to the cluster.
• Error::is_invalid_request_error If the request details are invalid. e.g., an invalid entity type was specified.
• Error::is_timeout_error If the request timed out before the describe could finish.

This operation is supported by brokers with version 2.6.0 or higher.

Arguments

• filter - the filter to apply to match entities
• options - the options to use

Returns

the DescribeClientQuotasResult containing the result

fn alter_client_quotas_with_options( &self, entries: &[ClientQuotaAlteration], options: AlterClientQuotasOptions, ) -> AlterClientQuotasResult

Alters client quota configurations with the specified alterations.

Alterations for a single entity are atomic, but across entities is not guaranteed. The resulting per-entity error code should be evaluated to resolve the success or failure of all updates.

The following errors can be anticipated when calling get() on the futures obtained from the returned AlterClientQuotasResult:

• Error::is_cluster_authorization_error If the authenticated user didn’t have alter access to the cluster.
• Error::is_invalid_request_error If the request details are invalid. e.g., a configuration key was specified more than once for an entity.
• Error::is_timeout_error If the request timed out before the alterations could finish. It cannot be guaranteed whether the update succeed or not.

This operation is supported by brokers with version 2.6.0 or higher.

Arguments

• entries - the alterations to perform
• options - The options to use when altering the quotas.

Returns

the AlterClientQuotasResult containing the result

fn describe_user_scram_credentials_with_users_options( &self, users: &[String], options: DescribeUserScramCredentialsOptions, ) -> DescribeUserScramCredentialsResult

Describe SASL/SCRAM credentials.

The following errors can be anticipated when calling get() on the futures from the returned DescribeUserScramCredentialsResult:

• Error::is_cluster_authorization_error If the authenticated user didn’t have describe access to the cluster.
• Error::is_resource_not_found_error If the user did not exist/had no SCRAM credentials.
• Error::is_duplicate_resource_error If the user was requested to be described more than once in the original request.
• Error::is_timeout_error If the request timed out before the describe operation could finish.

This operation is supported by brokers with version 2.7.0 or higher.

Arguments

• users - the users for which credentials are to be described; all users’ credentials are described if empty.
• options - The options to use when describing the credentials

Returns

The DescribeUserScramCredentialsResult.

fn alter_user_scram_credentials_with_options( &self, alterations: &[UserScramCredentialAlteration], options: AlterUserScramCredentialsOptions, ) -> AlterUserScramCredentialsResult

Alter SASL/SCRAM credentials.

The following errors can be anticipated when calling get() any of the futures from the returned AlterUserScramCredentialsResult:

• Error::is_not_controller_error If the request is not sent to the Controller broker.
• Error::is_cluster_authorization_error If the authenticated user didn’t have alter access to the cluster.
• Error::is_unsupported_by_authentication_error If the user authenticated with a delegation token.
• Error::is_unsupported_sasl_mechanism_error If the requested SCRAM mechanism is unrecognized or otherwise unsupported.
• Error::is_unacceptable_credential_error If the username is empty or the requested number of iterations is too small or too large.
• Error::is_timeout_error If the request timed out before the describe could finish.

This operation is supported by brokers with version 2.7.0 or higher.

Arguments

• alterations - the alterations to be applied
• options - The options to use when altering the credentials

Returns

The AlterUserScramCredentialsResult.

fn create_delegation_token_with_options( &self, options: CreateDelegationTokenOptions, ) -> CreateDelegationTokenResult

Create a Delegation Token.

This operation is supported by brokers with version 1.1.0 or higher.

The following errors can be anticipated when calling get() on the futures obtained from the delegationToken() method of the returned CreateDelegationTokenResult

• Error::is_unsupported_by_authentication_error If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
• Error::is_invalid_principal_type_error if the renewers principal type is not supported.
• Error::is_delegation_token_disabled_error if the delegation token feature is disabled.
• Error::is_timeout_error if the request was not completed in within the given CreateDelegationTokenOptions::timeout_ms.

Arguments

• options - The options to use when creating delegation token.

Returns

The CreateDelegationTokenResult.

fn renew_delegation_token_with_options( &self, hmac: &[u8], options: RenewDelegationTokenOptions, ) -> RenewDelegationTokenResult

Renew a Delegation Token.

This operation is supported by brokers with version 1.1.0 or higher.

The following errors can be anticipated when calling get() on the futures obtained from the expiryTimestamp() method of the returned RenewDelegationTokenResult

• Error::is_unsupported_by_authentication_error If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
• Error::is_delegation_token_disabled_error if the delegation token feature is disabled.
• Error::is_delegation_token_not_found_error if the delegation token is not found on server.
• Error::is_delegation_token_owner_mismatch_error if the authenticated user is not owner/renewer of the token.
• Error::is_delegation_token_expired_error if the delegation token is expired.
• Error::is_timeout_error if the request was not completed in within the given RenewDelegationTokenOptions::timeout_ms.

Arguments

• hmac - HMAC of the Delegation token
• options - The options to use when renewing delegation token.

Returns

The RenewDelegationTokenResult.

fn expire_delegation_token_with_options( &self, hmac: &[u8], options: ExpireDelegationTokenOptions, ) -> ExpireDelegationTokenResult

Expire a Delegation Token.

This operation is supported by brokers with version 1.1.0 or higher.

The following errors can be anticipated when calling get() on the futures obtained from the expiryTimestamp() method of the returned ExpireDelegationTokenResult

• Error::is_unsupported_by_authentication_error If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
• Error::is_delegation_token_disabled_error if the delegation token feature is disabled.
• Error::is_delegation_token_not_found_error if the delegation token is not found on server.
• Error::is_delegation_token_owner_mismatch_error if the authenticated user is not owner/renewer of the requested token.
• Error::is_delegation_token_expired_error if the delegation token is expired.
• Error::is_timeout_error if the request was not completed in within the given ExpireDelegationTokenOptions::timeout_ms.

Arguments

• hmac - HMAC of the Delegation token
• options - The options to use when expiring delegation token.

Returns

The ExpireDelegationTokenResult.

fn describe_delegation_token_with_options( &self, options: DescribeDelegationTokenOptions, ) -> DescribeDelegationTokenResult

Describe the Delegation Tokens.

This operation is supported by brokers with version 1.1.0 or higher.

The following errors can be anticipated when calling get() on the futures obtained from the delegationTokens() method of the returned DescribeDelegationTokenResult

• Error::is_unsupported_by_authentication_error If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
• Error::is_delegation_token_disabled_error if the delegation token feature is disabled.
• Error::is_timeout_error if the request was not completed in within the given DescribeDelegationTokenOptions::timeout_ms.

Arguments

• options - The options to use when describing delegation tokens.

Returns

The DescribeDelegationTokenResult.

fn describe_features_with_options( &self, options: DescribeFeaturesOptions, ) -> DescribeFeaturesResult

Describes finalized as well as supported features. The request is issued to any random broker.

The following errors can be anticipated when calling get() on the future from the returned DescribeFeaturesResult:

• Error::is_timeout_error If the request timed out before the describe operation could finish.

Arguments

• options - the options to use

Returns

the DescribeFeaturesResult containing the result

fn update_features_with_options( &self, feature_updates: &HashMap<String, FeatureUpdate>, options: UpdateFeaturesOptions, ) -> Result<UpdateFeaturesResult, Error>

Applies specified updates to finalized features. This operation is not transactional so some updates may succeed while the rest may fail.

The API takes in a map of finalized feature names to FeatureUpdate that needs to be applied. Each entry in the map specifies the finalized feature to be added or updated or deleted, along with the new max feature version level value. This request is issued only to the controller since the API is only served by the controller. The return value contains an error code for each supplied FeatureUpdate, and the code indicates if the update succeeded or failed in the controller.

• Downgrading a feature version level is not a common operation and should only be performed when necessary. It is permitted only if the FeatureUpdate specifies the upgradeType as either UpgradeType::SAFE_DOWNGRADE or UpgradeType::UNSAFE_DOWNGRADE.
• SAFE_DOWNGRADE: Allows downgrades that do not lead to metadata loss.
• UNSAFE_DOWNGRADE: Permits downgrades that might result in metadata loss.

Note that even with these settings, certain downgrades may still be rejected by the controller if they are considered unsafe or impossible.

• Deleting a finalized feature version is also not a common operation. To delete a feature, set the maxVersionLevel to zero and specify the upgradeType as either UpgradeType::SAFE_DOWNGRADE or UpgradeType::UNSAFE_DOWNGRADE.
• The UpgradeType::UPGRADE type cannot be used when the maxVersionLevel is zero. Attempting to do so will result in an Error::is_local_illegal_argument_error.

The following errors can be anticipated when calling get() on the futures obtained from the returned UpdateFeaturesResult:

• Error::is_cluster_authorization_error If the authenticated user didn’t have alter access to the cluster.
• Error::is_invalid_request_error If the request details are invalid. e.g., a non-existing finalized feature is attempted to be deleted or downgraded.
• Error::is_timeout_error If the request timed out before the updates could finish. It cannot be guaranteed whether the updates succeeded or not.
• Error::is_feature_update_failed_error This means there was an unexpected error encountered when the update was applied on the controller. There is no guarantee on whether the update succeeded or failed. The best way to find out is to issue a Admin::describe_features request.

This operation is supported by brokers with version 2.7.0 or higher.

Arguments

• feature_updates - the map of finalized feature name to FeatureUpdate
• options - the options to use

Returns

the UpdateFeaturesResult containing the result

Errors

• Error::is_local_illegal_argument_error if feature_updates is empty or any feature name is blank.

Translation notes

Java throws that IllegalArgumentException from the call; it is returned here as an Err.

fn close_with_timeout<'life0, 'async_trait>( &'life0 self, timeout: Duration, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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

Close the Admin client and release all associated resources.

The close operation has a grace period during which current operations will be allowed to complete, specified by the given duration. New operations will not be accepted during the grace period. Once the grace period is over, all operations that have not yet been completed will be aborted with a Error::is_timeout_error.

Arguments

• timeout - The time to use for the wait time.

Translation notes

Returns once timeout has elapsed, as Java’s timed join does; a background task still running then is left running.

Provided Methods

fn create_topics(&self, new_topics: &[NewTopic]) -> CreateTopicsResult

Create a batch of new topics with the default options.

This is a convenience method for Self::create_topics_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.10.1.0 or higher.

Arguments

• new_topics - The new topics to create.

Returns

The CreateTopicsResult.

fn delete_topics_with_string_slice( &self, topics: &[String], ) -> DeleteTopicsResult

This is a convenience method for Self::delete_topics_with_topic_collection_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.10.1.0 or higher.

Arguments

• topics - The topic names to delete.

Returns

The DeleteTopicsResult.

Translation notes

Java names the parameter of every deleteTopics overload topics, so each form is named after its type.

fn delete_topics_with_string_slice_options( &self, topics: &[String], options: DeleteTopicsOptions, ) -> DeleteTopicsResult

This is a convenience method for Self::delete_topics_with_topic_collection_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.10.1.0 or higher.

Arguments

• topics - The topic names to delete.
• options - The options to use when deleting the topics.

Returns

The DeleteTopicsResult.

fn delete_topics_with_topic_collection( &self, topics: TopicCollection, ) -> DeleteTopicsResult

This is a convenience method for Self::delete_topics_with_topic_collection_options with default options. See the overload for more details.

When using topic IDs, this operation is supported by brokers with inter-broker protocol 2.8 or higher. When using topic names, this operation is supported by brokers with version 0.10.1.0 or higher.

Arguments

• topics - The topics to delete.

Returns

The DeleteTopicsResult.

fn list_topics(&self) -> ListTopicsResult

List the topics available in the cluster with the default options.

This is a convenience method for Self::list_topics_with_options with default options. See the overload for more details.

Returns

The ListTopicsResult.

fn describe_topics_with_topic_names( &self, topic_names: &[String], ) -> DescribeTopicsResult

Describe some topics in the cluster, with the default options.

This is a convenience method for Self::describe_topics_with_topic_names_options with default options. See the overload for more details.

Arguments

• topic_names - The names of the topics to describe.

Returns

The DescribeTopicsResult.

fn describe_topics_with_topic_names_options( &self, topic_names: &[String], options: DescribeTopicsOptions, ) -> DescribeTopicsResult

Describe some topics in the cluster.

Arguments

• topic_names - The names of the topics to describe.
• options - The options to use when describing the topic.

Returns

The DescribeTopicsResult.

fn describe_topics_with_topics( &self, topics: TopicCollection, ) -> DescribeTopicsResult

This is a convenience method for Self::describe_topics_with_topics_options with default options. See the overload for more details.

When using topic IDs, this operation is supported by brokers with version 3.1.0 or higher.

Arguments

• topics - The topics to describe.

Returns

The DescribeTopicsResult.

fn create_partitions( &self, new_partitions: &HashMap<String, NewPartitions>, ) -> CreatePartitionsResult

Increase the number of partitions of the topics given as the keys of new_partitions according to the corresponding values. If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected.

This is a convenience method for Self::create_partitions_with_options with default options. See the overload for more details.

Arguments

• new_partitions - The topics which should have new partitions created, and corresponding parameters for the created partitions.

Returns

The CreatePartitionsResult.

fn delete_records( &self, records_to_delete: &HashMap<TopicPartition, RecordsToDelete>, ) -> DeleteRecordsResult

Delete records whose offset is smaller than the given offset of the corresponding partition.

This is a convenience method for Self::delete_records_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• records_to_delete - The topic partitions and related offsets from which records deletion starts.

Returns

The DeleteRecordsResult.

fn unregister_controller( &self, controller_id: i32, ) -> UnregisterControllerResult

Unregister a controller. This is a convenience method for Self::unregister_controller_with_options

Arguments

• controller_id - the controller id to unregister.

Returns

the UnregisterControllerResult containing the result

fn describe_producers( &self, partitions: &[TopicPartition], ) -> DescribeProducersResult

Describe producer state on a set of topic partitions. See Self::describe_producers_with_options for more details.

Arguments

• partitions - The set of partitions to query

Returns

The result

fn abort_transaction( &self, spec: AbortTransactionSpec, ) -> AbortTransactionResult

Forcefully abort a transaction which is open on a topic partition. See Self::abort_transaction_with_options for more details.

Arguments

• spec - The transaction specification including topic partition and producer details

Returns

The result

fn describe_transactions( &self, transactional_ids: &[String], ) -> DescribeTransactionsResult

Describe the state of a set of transactional IDs. See Self::describe_transactions_with_options for more details.

Arguments

• transactional_ids - The set of transactional IDs to query

Returns

The result

fn fence_producers(&self, transactional_ids: &[String]) -> FenceProducersResult

Fence out all active producers that use any of the provided transactional IDs, with the default options.

This is a convenience method for Self::fence_producers_with_options with default options. See the overload for more details.

Arguments

• transactional_ids - The IDs of the producers to fence.

Returns

The FenceProducersResult.

fn list_transactions(&self) -> ListTransactionsResult

List active transactions in the cluster. See Self::list_transactions_with_options for more details.

Returns

The result

fn force_terminate_transaction( &self, transactional_id: &str, ) -> TerminateTransactionResult

Force terminate a transaction for the given transactional ID with the default options.

This is a convenience method for Self::force_terminate_transaction_with_options with default options.

Arguments

• transactional_id - The ID of the transaction to terminate.

Returns

The TerminateTransactionResult.

fn describe_cluster(&self) -> DescribeClusterResult

Get information about the nodes in the cluster, using the default options.

This is a convenience method for Self::describe_cluster_with_options with default options. See the overload for more details.

Returns

The DescribeClusterResult.

fn describe_configs( &self, config_resources: &[ConfigResource], ) -> DescribeConfigsResult

Get the configuration for the specified resources with the default options.

This is a convenience method for Self::describe_configs_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• config_resources - The resources (topic and broker resource types are currently supported)

Returns

The DescribeConfigsResult

fn incremental_alter_configs( &self, configs: &HashMap<ConfigResource, Vec<AlterConfigOp>>, ) -> AlterConfigsResult

Incrementally updates the configuration for the specified resources with default options.

This is a convenience method for Self::incremental_alter_configs_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 2.3.0 or higher.

Arguments

• configs - The resources with their configs

Returns

The AlterConfigsResult

fn list_config_resources(&self) -> ListConfigResourcesResult

List all configuration resources available in the cluster with the default options.

This is a convenience method for Self::list_config_resources_with_options with default options. See the overload for more details.

Returns

The ListConfigurationResourcesResult.

fn describe_log_dirs(&self, brokers: &[i32]) -> DescribeLogDirsResult

Query the information of all log directories on the given set of brokers

This is a convenience method for Self::describe_log_dirs_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 1.0.0 or higher.

Arguments

• brokers - A list of brokers

Returns

The DescribeLogDirsResult

fn alter_replica_log_dirs( &self, replica_assignment: &HashMap<TopicPartitionReplica, String>, ) -> AlterReplicaLogDirsResult

Change the log directory for the specified replicas. If the replica does not exist on the broker, the result shows REPLICA_NOT_AVAILABLE for the given replica and the replica will be created in the given log directory on the broker when it is created later. If the replica already exists on the broker, the replica will be moved to the given log directory if it is not already there. For detailed result, inspect the returned AlterReplicaLogDirsResult instance.

This operation is not transactional so it may succeed for some replicas while fail for others.

This is a convenience method for Self::alter_replica_log_dirs_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 1.1.0 or higher.

Arguments

• replica_assignment - The replicas with their log directory absolute path

Returns

The AlterReplicaLogDirsResult

fn describe_replica_log_dirs( &self, replicas: &[TopicPartitionReplica], ) -> DescribeReplicaLogDirsResult

Query the replica log directory information for the specified replicas.

This is a convenience method for Self::describe_replica_log_dirs_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 1.0.0 or higher.

Arguments

• replicas - The replicas to query

Returns

The DescribeReplicaLogDirsResult

fn elect_leaders( &self, election_type: ElectionType, partitions: Option<HashSet<TopicPartition>>, ) -> ElectLeadersResult

Elect a replica as leader for topic partitions.

This is a convenience method for Self::elect_leaders_with_options with default options.

Arguments

• election_type - The type of election to conduct.
• partitions - The topics and partitions for which to conduct elections.

Returns

The ElectLeadersResult.

fn alter_partition_reassignments( &self, reassignments: &HashMap<TopicPartition, Option<NewPartitionReassignment>>, ) -> AlterPartitionReassignmentsResult

Change the reassignments for one or more partitions. Providing None will revert the reassignment for the associated partition. This is a convenience method for Self::alter_partition_reassignments_with_options with default options. See the overload for more details.

Arguments

• reassignments - The reassignments to add, modify, or remove. See NewPartitionReassignment.

Returns

The result.

fn list_partition_reassignments(&self) -> ListPartitionReassignmentsResult

List all of the current partition reassignments.

This is a convenience method for Self::list_partition_reassignments_with_options with default options. See the overload for more details.

Returns

The result.

fn list_partition_reassignments_with_partitions( &self, partitions: HashSet<TopicPartition>, ) -> ListPartitionReassignmentsResult

List the current reassignments for the given partitions.

This is a convenience method for Self::list_partition_reassignments_with_partitions_options with default options. See the overload for more details.

Arguments

• partitions - The topic partitions to list reassignments for.

Returns

The result.

fn list_partition_reassignments_with_options( &self, options: ListPartitionReassignmentsOptions, ) -> ListPartitionReassignmentsResult

List all of the current partition reassignments.

The following errors can be anticipated when calling get() on the futures obtained from the returned ListPartitionReassignmentsResult:

• Error::is_cluster_authorization_error If the authenticated user doesn’t have alter access to the cluster.
• Error::is_unknown_topic_or_partition_error If a given topic or partition does not exist.
• Error::is_timeout_error If the request timed out before the controller could list the current reassignments.

Arguments

• options - The options to use.

Returns

The result.

fn list_offsets( &self, topic_partition_offsets: &HashMap<TopicPartition, OffsetSpec>, ) -> ListOffsetsResult

List offset for the specified partitions and OffsetSpec. This operation enables to find the beginning offset, end offset as well as the offset matching a timestamp in partitions.

This is a convenience method for Self::list_offsets_with_options

Arguments

• topic_partition_offsets - The mapping from partition to the OffsetSpec to look up.

Returns

The ListOffsetsResult.

fn list_groups(&self) -> ListGroupsResult

List the groups available in the cluster with the default options.

This is a convenience method for Self::list_groups_with_options with default options. See the overload for more details.

Returns

The ListGroupsResult.

fn describe_consumer_groups( &self, group_ids: &[String], ) -> DescribeConsumerGroupsResult

Describe some consumer groups in the cluster, with the default options.

This is a convenience method for Self::describe_consumer_groups_with_options with default options. See the overload for more details.

Arguments

• group_ids - The IDs of the groups to describe.

Returns

The DescribeConsumerGroupsResult.

fn describe_classic_groups( &self, group_ids: &[String], ) -> DescribeClassicGroupsResult

Describe some classic groups in the cluster, with the default options.

This is a convenience method for Self::describe_classic_groups_with_options with default options. See the overload for more details.

Arguments

• group_ids - The IDs of the groups to describe.

Returns

The DescribeClassicGroupsResult.

fn list_consumer_group_offsets_with_group_id( &self, group_id: &str, ) -> ListConsumerGroupOffsetsResult

List the consumer group offsets available in the cluster with the default options.

This is a convenience method for Self::list_consumer_group_offsets_with_group_id_options to list offsets of all partitions of one group with default options.

Arguments

• group_id - The group to list the offsets of.

Returns

The ListConsumerGroupOffsetsResult.

fn list_consumer_group_offsets_with_group_id_options( &self, group_id: &str, options: ListConsumerGroupOffsetsOptions, ) -> ListConsumerGroupOffsetsResult

List the consumer group offsets available in the cluster.

Arguments

• group_id - The group to list the offsets of.
• options - The options to use when listing the consumer group offsets.

Returns

The ListConsumerGroupOffsetsResult

fn list_consumer_group_offsets_with_group_specs( &self, group_specs: &HashMap<String, ListConsumerGroupOffsetsSpec>, ) -> ListConsumerGroupOffsetsResult

List the consumer group offsets available in the cluster for the specified groups with the default options.

This is a convenience method for Self::list_consumer_group_offsets_with_group_specs_options with default options.

Arguments

• group_specs - Map of consumer group ids to a spec that specifies the topic partitions of the group to list offsets for.

Returns

The ListConsumerGroupOffsetsResult.

fn alter_consumer_group_offsets( &self, group_id: &str, offsets: &HashMap<TopicPartition, OffsetAndMetadata>, ) -> AlterConsumerGroupOffsetsResult

Alters offsets for the specified group. In order to succeed, the group must be empty.

This is a convenience method for Self::alter_consumer_group_offsets_with_options with default options. See the overload for more details.

Arguments

• group_id - The group for which to alter offsets.
• offsets - A map of offsets by partition with associated metadata.

Returns

The AlterOffsetsResult.

fn delete_consumer_group_offsets( &self, group_id: &str, partitions: &HashSet<TopicPartition>, ) -> DeleteConsumerGroupOffsetsResult

Delete committed offsets for a set of partitions in a consumer group with the default options. This will succeed at the partition level only if the group is not actively subscribed to the corresponding topic.

Arguments

• group_id - The group to delete the offsets of.
• partitions - The partitions whose committed offsets to delete.

Returns

The DeleteConsumerGroupOffsetsResult.

fn delete_consumer_groups( &self, group_ids: &[String], ) -> DeleteConsumerGroupsResult

Delete consumer groups from the cluster with the default options.

Arguments

• group_ids - The ids of the groups to delete.

Returns

The DeleteConsumerGroupResult.

fn create_acls(&self, acls: &[AclBinding]) -> CreateAclsResult

This is a convenience method for Self::create_acls_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• acls - The ACLs to create

Returns

The CreateAclsResult.

fn describe_acls(&self, filter: &AclBindingFilter) -> DescribeAclsResult

This is a convenience method for Self::describe_acls_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• filter - The filter to use.

Returns

The DescribeAclsResult.

fn delete_acls(&self, filters: &[AclBindingFilter]) -> DeleteAclsResult

This is a convenience method for Self::delete_acls_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 0.11.0.0 or higher.

Arguments

• filters - The filters to use.

Returns

The DeleteAclsResult.

fn describe_client_quotas( &self, filter: &ClientQuotaFilter, ) -> DescribeClientQuotasResult

Describes all entities matching the provided filter that have at least one client quota configuration value defined.

This is a convenience method for Self::describe_client_quotas_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 2.6.0 or higher.

Arguments

• filter - the filter to apply to match entities

Returns

the DescribeClientQuotasResult containing the result

fn alter_client_quotas( &self, entries: &[ClientQuotaAlteration], ) -> AlterClientQuotasResult

Alters client quota configurations with the specified alterations.

This is a convenience method for Self::alter_client_quotas_with_options with default options. See the overload for more details.

This operation is supported by brokers with version 2.6.0 or higher.

Arguments

• entries - the alterations to perform

Returns

the AlterClientQuotasResult containing the result

fn describe_user_scram_credentials(&self) -> DescribeUserScramCredentialsResult

Describe all SASL/SCRAM credentials.

This is a convenience method for Self::describe_user_scram_credentials_with_users_options

Returns

The DescribeUserScramCredentialsResult.

fn describe_user_scram_credentials_with_users( &self, users: &[String], ) -> DescribeUserScramCredentialsResult

Describe SASL/SCRAM credentials for the given users.

This is a convenience method for Self::describe_user_scram_credentials_with_users_options

Arguments

• users - the users for which credentials are to be described; all users’ credentials are described if empty.

Returns

The DescribeUserScramCredentialsResult.

fn alter_user_scram_credentials( &self, alterations: &[UserScramCredentialAlteration], ) -> AlterUserScramCredentialsResult

Alter SASL/SCRAM credentials for the given users.

This is a convenience method for Self::alter_user_scram_credentials_with_options

Arguments

• alterations - the alterations to be applied

Returns

The AlterUserScramCredentialsResult.

fn create_delegation_token(&self) -> CreateDelegationTokenResult

Create a Delegation Token.

This is a convenience method for Self::create_delegation_token_with_options with default options. See the overload for more details.

Returns

The CreateDelegationTokenResult.

fn renew_delegation_token(&self, hmac: &[u8]) -> RenewDelegationTokenResult

Renew a Delegation Token.

This is a convenience method for Self::renew_delegation_token_with_options with default options. See the overload for more details.

Arguments

• hmac - HMAC of the Delegation token

Returns

The RenewDelegationTokenResult.

fn expire_delegation_token(&self, hmac: &[u8]) -> ExpireDelegationTokenResult

Expire a Delegation Token.

This is a convenience method for Self::expire_delegation_token_with_options with default options. This will expire the token immediately. See the overload for more details.

Arguments

• hmac - HMAC of the Delegation token

Returns

The ExpireDelegationTokenResult.

fn describe_delegation_token(&self) -> DescribeDelegationTokenResult

Describe the Delegation Tokens.

This is a convenience method for Self::describe_delegation_token_with_options with default options. This will return all the user owned tokens and tokens where user have Describe permission. See the overload for more details.

Returns

The DescribeDelegationTokenResult.

fn describe_features(&self) -> DescribeFeaturesResult

Describes finalized as well as supported features.

This is a convenience method for Self::describe_features_with_options with default options. See the overload for more details.

Returns

the DescribeFeaturesResult containing the result

fn update_features( &self, feature_updates: &HashMap<String, FeatureUpdate>, ) -> Result<UpdateFeaturesResult, Error>

Applies specified updates to finalized features.

This is a convenience method for Self::update_features_with_options with default options. See the overload for more details.

Arguments

• feature_updates - the map of finalized feature name to FeatureUpdate

Returns

the UpdateFeaturesResult containing the result

Errors

• Error::is_local_illegal_argument_error if feature_updates is empty or any feature name is blank.

fn close<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

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

Close the Admin and release all associated resources.

See Self::close_with_timeout

Implementors

impl Admin for MockAdminClient

impl Admin for KafkaAdminClient