Struct AdminClient 

pub struct AdminClient { /* private fields */ }

Short-lived admin client that targets one cluster’s controller. It can negotiate TLS/SASL through AdminClient::connect_secured.

As Kafka’s admin client with enable.metrics.push on (the default), it pushes the client metrics that a broker’s client metrics subscription names (KIP-714) on its bootstrap connection, and its connections count in the admin-client-metrics network metrics. AdminClientConfig::enable_metrics_push turns this off. The constructors without an AdminClientConfig keep Kafka’s default. close sends the terminating push and waits for it; dropping the client sends it in the background.

Implementations

impl AdminClient

pub async fn unregister_controller( &mut self, controller_id: i32, ) -> Result<(), AdminError>

Unregisters one controller from the active KRaft controller, with an UnregisterController request (API key 94).

This is Kafka’s Admin.unregisterController. Like KafkaAdminClient.unregisterBroker, it goes to a broker of a broker connection or to the active controller of a controller connection. It retries REQUEST_TIMED_OUT (7) with the retry backoff until the call deadline, and fails at once on every other code: NOT_CONTROLLER (41), INVALID_REQUEST (42) for a controller in the voter set, and CONTROLLER_ID_NOT_REGISTERED (136).

Errors

Returns a transport, protocol, or broker error. At the deadline it gives REQUEST_TIMED_OUT (7).

impl AdminClient

pub async fn describe_cluster( &self, options: DescribeClusterOptions, ) -> Result<ClusterDescription, AdminError>

Describes the cluster: its ID, its nodes, the node that serves as controller, and optionally the operations that the caller may perform on it, as Kafka’s describeCluster operation does.

The client sends DescribeCluster on its connection. Through broker bootstrap it asks for broker endpoints; through controller bootstrap (KIP-919) it asks for controller endpoints with v1 or higher, as Kafka’s KafkaAdminClient does. include_fenced_brokers needs v2, as Kafka’s builder refuses a lower version. The call stops at Kafka’s default default.api.timeout.ms (60 s).

Errors

Returns AdminError::Broker for a top-level error, such as CLUSTER_AUTHORIZATION_FAILED (31), and with REQUEST_TIMED_OUT (7) at the deadline. Returns AdminError::Transport when the connection fails, or when the peer does not support the DescribeCluster version that the options need.

impl AdminClient

pub async fn list_config_resources( &self, types: &BTreeSet<ConfigResourceType>, ) -> Result<Vec<ConfigResource>, AdminError>

Lists the config resources of types, as Kafka’s listConfigResources operation does. An empty set asks for every type the broker supports.

The client sends one ListConfigResources request on its connection. Exactly {ClientMetrics} may go as v0, as Kafka’s builder allows; every other set needs v1 (KIP-1142).

Errors

Returns AdminError::Broker for a top-level error, such as CLUSTER_AUTHORIZATION_FAILED (31) or UNSUPPORTED_VERSION (35) for a type the broker does not list, and with REQUEST_TIMED_OUT (7) at default.api.timeout.ms (60 s). Returns AdminError::Transport when the connection fails, or when the broker supports only v0 and types is not {ClientMetrics}.

pub async fn list_client_metrics_resources( &self, ) -> Result<Vec<String>, AdminError>

Lists the names of the client-metrics subscriptions (KIP-714), as Kafka’s listClientMetricsResources operation does: a ListConfigResources request for {ClientMetrics}, keeping only the resources of that type.

Errors

See Self::list_config_resources.

impl AdminClient

pub async fn list_consumer_groups( &self, options: &ListConsumerGroupsOptions, ) -> Result<ListConsumerGroupsResult, AdminError>

Lists the consumer groups of the whole cluster, as Apache Kafka’s listConsumerGroups does.

The call sends ListGroups to every broker with the state and type filters of options, and keeps each group whose protocol type is consumer or empty. See AdminClient::list_groups for the fan-out, the retries, the per-broker errors and the version that each filter needs.

Errors

Returns an error when the Metadata request fails, or when the metadata names no broker after the timeout. A failure on one broker is an entry in ListConsumerGroupsResult::errors.

impl AdminClient

pub async fn elect_leaders( &mut self, election_type: ElectionType, partitions: Option<&[(String, i32)]>, ) -> Result<ElectionResults, AdminError>

Elects a leader for each partition of partitions, or of every partition of the cluster for None, as Kafka’s electLeaders operation does.

The client sends ElectLeaders to the controller connection, as Kafka’s KafkaAdminClient sends it with a ControllerNodeProvider, with the time left before the call deadline as its timeout_ms. ElectionType::Unclean needs ElectLeaders v1, so the client negotiates v1 or higher for it, as Kafka’s builder refuses v0.

NOT_CONTROLLER (41), top-level or on any partition, makes the client find the controller again and resend the request with the backoff, until Kafka’s default default.api.timeout.ms (60 s).

The result has an entry for each partition that the controller answered for. A partition that needed no election gets ELECTION_NOT_NEEDED (84), as Kafka gives an ElectionNotNeededException.

Errors

Returns AdminError::Broker for a top-level error, such as CLUSTER_AUTHORIZATION_FAILED (31), and with REQUEST_TIMED_OUT (7) at the deadline. Returns AdminError::Transport when the connection fails, or when the controller does not support the ElectLeaders version that election_type needs.

impl AdminClient

pub async fn describe_features( &self, options: DescribeFeaturesOptions, ) -> Result<FeatureMetadata, AdminError>

Returns the supported and cluster-finalized feature ranges of one node, as Kafka’s describeFeatures.

Errors

Returns a transport, protocol, or broker error. With a node id, a node that the metadata does not name, or that does not answer, is retried until the deadline and then gives REQUEST_TIMED_OUT (7).

pub async fn update_features( &mut self, updates: &BTreeMap<String, FeatureUpdate>, options: UpdateFeaturesOptions, ) -> Result<UpdateFeaturesResults, AdminError>

Applies finalized-feature updates through UpdateFeatures on the active controller, as Kafka’s updateFeatures.

Every feature of updates gets a result. A top-level error of the controller fails every feature with it. A v2 response with no per-feature results and no error succeeds every feature, and a feature that a v0/v1 response leaves out fails with UNKNOWN_SERVER_ERROR, as Kafka does.

Errors

Returns AdminError::InvalidArgument, and sends nothing, for no updates or a blank feature name, as Kafka throws IllegalArgumentException. Returns [krabka_client_core::ClientError::IncompatibleVersion] before sending when a downgrade or UpdateFeaturesOptions::validate_only meets a controller with only v0, as Kafka’s request serializer throws UnsupportedVersionException. A NOT_CONTROLLER answer is retried after controller discovery until the deadline, then gives REQUEST_TIMED_OUT (7). Transport errors are returned as they are.

impl AdminClient

pub async fn delete_consumer_groups(&self, groups: &[&str]) -> GroupDeletions

Deletes empty consumer groups with DeleteGroups, as Apache Kafka’s deleteConsumerGroups does with its DeleteConsumerGroupsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request and sends one DeleteGroups request to each coordinator with the groups that it coordinates. COORDINATOR_LOAD_IN_PROGRESS (14) sends a group to the same coordinator again, and COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16) find its coordinator again. A lost connection finds the coordinator again. The call waits with Kafka’s backoff between rounds and stops at default.api.timeout.ms (60 s), where each pending group gets REQUEST_TIMED_OUT (7).

Errors

The call itself does not fail. Each group gets its own KafkaError, such as NON_EMPTY_GROUP (68), GROUP_ID_NOT_FOUND (69) or REQUEST_TIMED_OUT (7) at the deadline.

pub async fn delete_share_groups(&self, groups: &[&str]) -> GroupDeletions

Deletes empty share groups. Apache Kafka’s deleteShareGroups uses a DeleteShareGroupsHandler that builds the same DeleteGroups request as deleteConsumerGroups.

Errors

See AdminClient::delete_consumer_groups.

pub async fn delete_streams_groups(&self, groups: &[&str]) -> GroupDeletions

Deletes empty streams groups. Apache Kafka’s deleteStreamsGroups delegates to deleteConsumerGroups.

Errors

See AdminClient::delete_consumer_groups.

impl AdminClient

pub async fn describe_consumer_groups( &self, groups: &[&str], options: DescribeGroupsOptions, ) -> GroupDescriptions<ConsumerGroupDescription>

Describes consumer groups, classic and KIP-848, as Apache Kafka’s describeConsumerGroups does with its DescribeConsumerGroupsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request, and sends one ConsumerGroupDescribe request to each coordinator with the groups that it coordinates. A group that the answer gives GROUP_ID_NOT_FOUND (69) or UNSUPPORTED_VERSION (35), and every group of a coordinator that does not support ConsumerGroupDescribe, is described with DescribeGroups instead, as Kafka’s handler falls back for a classic group. Kafka sends that DescribeGroups request in the next round of its driver. This call sends it in the same round, to the same coordinator.

COORDINATOR_LOAD_IN_PROGRESS (14) sends a group to the same coordinator again, and COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16) find its coordinator again. A lost connection finds the coordinator again. The call waits with Kafka’s backoff between rounds and stops at default.api.timeout.ms (60 s), where each pending group gets REQUEST_TIMED_OUT (7).

Errors

The call itself does not fail. Each group gets its own KafkaError: the error code of the answer, GROUP_ID_NOT_FOUND with the message of the ConsumerGroupDescribe answer for a group that does not exist, UNKNOWN_SERVER_ERROR (-1) for a classic group that is not a consumer group, or REQUEST_TIMED_OUT (7) at the deadline.

pub async fn describe_classic_groups( &self, groups: &[&str], options: DescribeGroupsOptions, ) -> GroupDescriptions<ClassicGroupDescription>

Describes classic groups with DescribeGroups, as Apache Kafka’s describeClassicGroups does with its DescribeClassicGroupsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request and sends one DescribeGroups request to each coordinator. The coordinator codes retry as AdminClient::describe_consumer_groups describes. Every other code is the final error of its group, with no message, as the handler uses error.message().

Errors

The call itself does not fail. Each group gets its own KafkaError.

pub async fn describe_share_groups( &self, groups: &[&str], options: DescribeGroupsOptions, ) -> GroupDescriptions<ShareGroupDescription>

Describes share groups with ShareGroupDescribe, as Apache Kafka’s describeShareGroups does with its DescribeShareGroupsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request and sends one ShareGroupDescribe request to each coordinator. The coordinator codes retry as AdminClient::describe_consumer_groups describes. Every other code, such as GROUP_ID_NOT_FOUND, is the final error of its group, with the message of the answer.

Errors

The call itself does not fail. Each group gets its own KafkaError.

pub async fn describe_streams_groups( &self, groups: &[&str], options: DescribeGroupsOptions, ) -> GroupDescriptions<StreamsGroupDescription>

Describes streams groups with StreamsGroupDescribe v0, as Apache Kafka’s describeStreamsGroups does with its DescribeStreamsGroupsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request and sends one StreamsGroupDescribe request to each coordinator. The coordinator codes retry as AdminClient::describe_consumer_groups describes. Every other code is the final error of its group, with the message of the answer. A group whose answer has no topology gets UNKNOWN_SERVER_ERROR (-1), as the handler fails it with an IllegalStateException.

Errors

The call itself does not fail. Each group gets its own KafkaError.

impl AdminClient

pub async fn alter_consumer_group_offsets( &mut self, group: &str, offsets: &GroupOffsets, ) -> Result<Vec<ConsumerGroupOffsetOutcome>, AdminError>

Commits explicit offsets for an inactive consumer group.

The request names each topic, so the client negotiates OffsetCommit v9 or lower, as Apache Kafka’s AlterConsumerGroupOffsetsHandler does.

The call retries the partition error codes as Apache Kafka’s AlterConsumerGroupOffsetsHandler.handleError does. One retriable code in the response makes the call send all the offsets again:

• COORDINATOR_LOAD_IN_PROGRESS (14) and REBALANCE_IN_PROGRESS (27): send the request again to the same coordinator.
• COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16): find the coordinator again, then send the request again.

A FindCoordinator answer of 14 or 15 also makes the call find the coordinator again. A failed or lost connection to the coordinator, including a TLS or SASL handshake that stops before the broker gives a verdict, makes the call find the coordinator again, as Kafka’s AdminApiDriver.onFailure does. The call waits between attempts with Kafka’s backoff, and it stops when Kafka’s default default.api.timeout.ms (60 s) elapses. A call whose retries are unresolved at the deadline, or whose attempt is still running, fails with ClientError::Timeout, as Kafka gives a TimeoutException.

Errors

Returns an error when encoding, transport, or response handling fails. Returns ClientError::Server when FindCoordinator answers with an error code that Kafka does not retry. Returns ClientError::Timeout at the deadline. Returns ClientError::IncompatibleVersion when the coordinator does not support OffsetCommit v9 or lower.

pub async fn list_groups( &self, options: &ListGroupsOptions, ) -> Result<ListGroupsResult, AdminError>

Lists the groups of the whole cluster, as Apache Kafka’s KafkaAdminClient.listGroups does.

A broker lists only the groups that it coordinates. The call first sends Metadata to find every broker. It then sends ListGroups to each broker at the same time, on a new connection, and merges the answers:

• The result has one GroupListing for each group id. A group that two brokers list appears once.
• A broker that fails gives one ListGroupsError. The groups of the other brokers stay in ListGroupsResult::valid.
• COORDINATOR_LOAD_IN_PROGRESS (14) and COORDINATOR_NOT_AVAILABLE (15) make the call send ListGroups to that broker again. A failed connection or a lost connection also makes the call try again. A rejected TLS or SASL authentication does not. It is an error for that broker at once.
• Kafka’s default default.api.timeout.ms (60 s) is the deadline of each broker. Each attempt on a broker (connection, TLS and SASL handshakes, and request) gets the time that remains. A broker that has not answered at the deadline gives REQUEST_TIMED_OUT (7), with the last error in the message.
• A filter that the broker version does not support gives UNSUPPORTED_VERSION (35) for that broker. states_filter needs ListGroups v4 (KIP-518) and types_filter needs v5 (KIP-848). Below v5, the call omits a type filter that holds Classic, and Consumer at most, as Kafka’s ListGroupsRequest.Builder.build does.
• ListGroupsOptions::protocol_types filters on the client.

Errors

Returns an error when the Metadata request fails, or when the metadata names no broker after the timeout. A failure on one broker is an entry in ListGroupsResult::errors, not an error of the call.

pub async fn list_consumer_group_offsets( &mut self, group: &str, ) -> Result<GroupOffsets, AdminError>

Returns (topic, partition) → committed_offset for the named group.

The call requests all topics and partitions (topics: None). It skips a partition that has an error code, and a partition with a committed offset below 0, which means no committed offset.

The response names each topic, so the client negotiates OffsetFetch v2 to v9, as Apache Kafka’s ListConsumerGroupOffsetsHandler does.

The call retries the coordinator error codes as Apache Kafka’s ListConsumerGroupOffsetsHandler.handleGroupError does:

• COORDINATOR_LOAD_IN_PROGRESS (14): send the request again to the same coordinator.
• COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16): find the coordinator again, then send the request again.

A FindCoordinator answer of 14 or 15 also makes the call find the coordinator again, as Kafka’s CoordinatorStrategy.handleError does. A failed or lost connection to the coordinator does the same, as AdminApiDriver.onFailure does. The call waits between attempts, and it stops when Kafka’s default default.api.timeout.ms (60 s) elapses. A call whose retries are unresolved at the deadline, or whose attempt is still running, fails with ClientError::Timeout.

Errors

Returns an error when encoding, transport, or response handling fails. Returns AdminError::Broker when the coordinator answers with a group error code that Kafka does not retry. Returns ClientError::Server when FindCoordinator answers with an error code that Kafka does not retry. Returns ClientError::Timeout at the deadline. Returns ClientError::IncompatibleVersion when the coordinator does not support OffsetFetch v2 to v9.

pub async fn delete_consumer_group_offsets( &mut self, group: &str, partitions: &[(String, i32)], ) -> Result<Vec<ConsumerGroupOffsetOutcome>, AdminError>

Deletes committed offsets of an inactive consumer group for the named partitions with OffsetDelete, as Apache Kafka’s deleteConsumerGroupOffsets does. Routes to the group’s coordinator with the same retry policy as AdminClient::alter_consumer_group_offsets.

Errors

Returns an error when encoding, transport, or response handling fails. Returns AdminError::Broker when the coordinator’s top-level answer carries a group error code that Kafka does not retry, such as GROUP_ID_NOT_FOUND or NON_EMPTY_GROUP.

pub async fn delete_streams_group_offsets( &mut self, group: &str, partitions: &[(String, i32)], ) -> Result<Vec<ConsumerGroupOffsetOutcome>, AdminError>

Deletes committed offsets of an inactive streams group. Apache Kafka’s KafkaAdminClient.deleteStreamsGroupOffsets delegates to deleteConsumerGroupOffsets verbatim, so this does too.

Errors

See AdminClient::delete_consumer_group_offsets.

pub async fn remove_members_from_consumer_group( &mut self, group: &str, members: &[MemberToRemove], options: &RemoveMembersOptions, ) -> Result<RemovedMembers, AdminError>

Removes members from a consumer group without waiting for their session timeout, with LeaveGroup, as Apache Kafka’s removeMembersFromConsumerGroup does. Routes to the group’s coordinator with the same retry policy as AdminClient::alter_consumer_group_offsets.

Each member is named by its group.instance.id, as Kafka’s MemberToRemove names it, and is sent with the member id UNKNOWN_MEMBER_ID (empty). An empty members removes every member of the group, as Kafka’s removeAll mode does: the call first describes the group with AdminClient::describe_consumer_groups and names each member by its instance id, or by its member id when it has none.

Each member carries options.reason, cut to 255 characters, or Kafka’s default "member was removed by an admin" when the reason is None or empty. LeaveGroup’s member list needs v3, so the client negotiates v3 or higher, as Kafka’s LeaveGroupRequest.Builder does.

The result maps each member of the LeaveGroup answer, by member id and instance id, to its error, as Kafka’s RemoveMembersFromConsumerGroupHandler.handleResponse does.

Errors

Returns an error when encoding, transport, or response handling fails. Returns AdminError::Broker when the coordinator’s top-level answer carries a group error code that Kafka does not retry, or when the removeAll describe fails for the group. Returns ClientError::IncompatibleVersion when the coordinator does not support LeaveGroup v3 or higher.

pub async fn alter_share_group_offsets( &mut self, group: &str, offsets: &GroupOffsets, ) -> Result<Vec<ConsumerGroupOffsetOutcome>, AdminError>

Commits explicit start offsets for a share group’s partitions with AlterShareGroupOffsets, as Apache Kafka’s alterShareGroupOffsets does. Routes to the group’s coordinator with the same retry policy as AdminClient::alter_consumer_group_offsets.

A top-level COORDINATOR_LOAD_IN_PROGRESS (14) or REBALANCE_IN_PROGRESS (27) sends the request again to the same coordinator, and COORDINATOR_NOT_AVAILABLE (15) or NOT_COORDINATOR (16) finds the coordinator again. Every other top-level code, such as NON_EMPTY_GROUP (68), is the error of each requested partition, as Kafka’s AlterShareGroupOffsetsHandler.handleError puts it in the partition results.

Errors

Returns an error when the coordinator lookup, the connection, or the request fails, or at the deadline.

pub async fn alter_streams_group_offsets( &mut self, group: &str, offsets: &GroupOffsets, ) -> Result<Vec<ConsumerGroupOffsetOutcome>, AdminError>

Commits explicit start offsets for a streams group’s partitions. Apache Kafka’s KafkaAdminClient.alterStreamsGroupOffsets delegates to alterConsumerGroupOffsets verbatim, so this does too.

Errors

See AdminClient::alter_consumer_group_offsets.

pub async fn list_streams_group_offsets( &mut self, group: &str, ) -> Result<GroupOffsets, AdminError>

Reads committed start offsets for a streams group’s partitions. Apache Kafka’s KafkaAdminClient.listStreamsGroupOffsets delegates to listConsumerGroupOffsets verbatim, so this does too.

Errors

See AdminClient::list_consumer_group_offsets.

pub async fn delete_share_group_offsets( &mut self, group: &str, topics: &[String], ) -> Result<Vec<(String, Option<KafkaError>)>, AdminError>

Deletes a share group’s committed offsets for the named topics with DeleteShareGroupOffsets, as Apache Kafka’s deleteShareGroupOffsets does. Routes to the group’s coordinator with the same retry policy as AdminClient::alter_consumer_group_offsets: a top-level COORDINATOR_LOAD_IN_PROGRESS (14) or REBALANCE_IN_PROGRESS (27) retries on the same coordinator, and COORDINATOR_NOT_AVAILABLE (15) or NOT_COORDINATOR (16) finds the coordinator again.

Errors

Returns an error when encoding, transport, or response handling fails. Returns AdminError::Broker when the coordinator’s top-level answer carries a group error code that Kafka does not retry.

impl AdminClient

pub async fn alter_replica_log_dirs( &mut self, assignments: &BTreeMap<TopicPartitionReplica, String>, ) -> BTreeMap<TopicPartitionReplica, BrokerResult<()>>

AlterReplicaLogDirs (KIP-113): moves replicas between the log.dirs of their brokers, as Kafka’s KafkaAdminClient.alterReplicaLogDirs does.

assignments maps each replica to the absolute path of its target log dir. The call groups the replicas by broker id and sends one request to each broker, all at the same time. The result has one entry for each replica:

• The error code of the broker for that partition, Ok(()) for 0.
• UNKNOWN_SERVER_ERROR (-1) when the response of the broker has no result for the replica (completeUnrealizedFutures).
• The error of the broker call for every replica of a broker that fails. The call finds each broker by id in fresh metadata for each connection attempt, so a broker that is missing from the metadata or that moved to a new address is found again. A missing broker and a failed or lost connection are tried again with Kafka’s backoff until default.api.timeout.ms (60 s), and then give REQUEST_TIMED_OUT (7). A slow or missing broker does not delay the other brokers.

pub async fn describe_log_dirs( &mut self, brokers: &[i32], filter: Option<&BTreeMap<String, Vec<i32>>>, ) -> BTreeMap<i32, BrokerResult<Vec<LogDirInfo>>>

DescribeLogDirs (KIP-113): lists every configured log.dir of each broker in brokers, with the partitions each one holds, as Kafka’s KafkaAdminClient.describeLogDirs does.

Pass None to fetch all partitions, as Kafka does. Pass Some with a topic to partitions filter to narrow the result. An empty inner vec means all partitions of that topic.

The result has one entry for each broker id, and the brokers are asked at the same time. A broker that answers with no log dir gives its top-level error code, or CLUSTER_AUTHORIZATION_FAILED (31) when it has none, as Kafka does. The call finds each broker by id in fresh metadata for each connection attempt. A missing broker and a failed or lost connection are tried again with Kafka’s backoff until default.api.timeout.ms (60 s), and then give REQUEST_TIMED_OUT (7).

pub async fn describe_replica_log_dirs( &mut self, replicas: &[TopicPartitionReplica], ) -> BTreeMap<TopicPartitionReplica, BrokerResult<ReplicaLogDirInfo>>

DescribeLogDirs for single replicas: finds the log dir of each replica of replicas, and of its future replica when one is being moved, as Kafka’s KafkaAdminClient.describeReplicaLogDirs does.

The call groups the replicas by broker id and sends each broker one DescribeLogDirs request that names only its partitions, all at the same time. Each broker is found and retried as in Self::describe_log_dirs. The result has one entry for each replica, and the brokers complete their replicas in the order they answer, as Kafka completes the replica futures:

• The log dirs that the broker reports for the partition. A log dir with KAFKA_STORAGE_ERROR (56) is offline and skipped, as Kafka does, so a replica that lives only there has neither log dir. A replica that the answer does not name has neither log dir.
• The top-level error code of the answer is not read, as Kafka’s describeReplicaLogDirs does not read it: a DescribeLogDirs v3+ answer with CLUSTER_AUTHORIZATION_FAILED (31) and no log dir gives each replica of that broker neither log dir.
• A log dir with any other error fails the call, as Kafka’s handleFailure with an IllegalStateException completes every replica future of the call that is still pending: the replicas of that broker and of every broker that has not answered yet get that error. The replicas of brokers that answered before keep their log dirs.
• A broker call that fails, with REQUEST_TIMED_OUT (7) at default.api.timeout.ms or another error, fails every pending replica of the call in the same way.

The call returns once no replica is pending; it does not wait for the brokers that have not answered after a failure.

impl AdminClient

pub async fn list_offsets( &self, specs: &BTreeMap<(String, i32), OffsetSpec>, isolation_level: IsolationLevel, ) -> BTreeMap<(String, i32), Result<ListedOffset, KafkaError>>

Looks up the offset that specs names for each partition, as Kafka’s listOffsets operation does.

The client finds the leader of each partition with Metadata and sends one ListOffsets request to each leader, all at the same time, as Kafka’s ListOffsetsHandler with its PartitionLeaderStrategy does. The request asks as a consumer (replica_id -1) with isolation_level, and carries the call timeout as its timeout_ms. The result has one entry for each partition of specs.

A partition goes back to the leader lookup, and the call finds its leader again after the backoff, when the lookup names no leader yet, when the connection to its leader fails or is lost, and when the leader answers a retriable code, such as NOT_LEADER_OR_FOLLOWER (6) or LEADER_NOT_AVAILABLE (5). The call stops at Kafka’s default default.api.timeout.ms (60 s), where each unresolved partition gets REQUEST_TIMED_OUT (7).

A spec that the leader’s ListOffsets versions cannot answer, such as OffsetSpec::MaxTimestamp below v7, fails with UNSUPPORTED_VERSION (35) for its partition only, as Kafka’s ListOffsetsHandler.handleUnsupportedError does.

Errors

The call itself does not fail. Each partition gets its own KafkaError in the result, such as the final lookup or ListOffsets error of that partition.

impl AdminClient

pub async fn add_raft_voter( &mut self, cluster_id: Option<&str>, voter_id: i32, voter_directory_id: Uuid, endpoints: &[RaftVoterEndpoint], ) -> Result<(), AdminError>

Adds a voter to the metadata quorum (KIP-853), as Kafka’s addRaftVoter operation does.

The client sends AddRaftVoter on its connection: the active controller through controller bootstrap (KIP-919), or a broker, which forwards it, as Kafka’s LeastLoadedBrokerOrActiveKController picks. cluster_id is Kafka’s AddRaftVoterOptions.clusterId; None sends a null cluster ID, which the controller does not check. The request carries the time left before the call deadline as its timeout_ms.

NOT_CONTROLLER (41) makes the client find the controller again and resend the request with the backoff until default.api.timeout.ms (60 s), as Kafka’s handleNotControllerError does.

Errors

Returns AdminError::Broker with the controller’s error, such as DUPLICATE_VOTER (126), and with REQUEST_TIMED_OUT (7) at the deadline, or a transport error.

impl AdminClient

pub async fn describe_client_quotas( &self, filter: &ClientQuotaFilter, ) -> Result<ClientQuotas, AdminError>

Describes the quotas of every entity that filter matches, as Kafka’s describeClientQuotas operation does.

The client sends one DescribeClientQuotas request on its connection and stops at default.api.timeout.ms (60 s).

Errors

Returns AdminError::Broker for a top-level error, such as CLUSTER_AUTHORIZATION_FAILED (31) or INVALID_REQUEST (42) for a filter the broker refuses, and with REQUEST_TIMED_OUT (7) at the deadline, or a transport error.

pub async fn alter_client_quotas( &self, alterations: &[ClientQuotaAlteration], validate_only: bool, ) -> Result<BTreeMap<ClientQuotaEntity, Result<(), KafkaError>>, AdminError>

Changes the quotas of each entity of alterations, as Kafka’s alterClientQuotas operation does. With validate_only the broker validates the changes and applies none.

The result has one entry for each entity: Ok(()), or the error the broker gave for it, such as INVALID_REQUEST (42) for an entity type combination it does not accept.

Errors

Returns a transport error, and AdminError::Broker with REQUEST_TIMED_OUT (7) at default.api.timeout.ms (60 s).

impl AdminClient

pub async fn list_share_group_offsets( &self, specs: &BTreeMap<String, ListShareGroupOffsetsSpec>, ) -> BTreeMap<String, Result<ShareGroupOffsets, KafkaError>>

Lists the start offsets of share groups with DescribeShareGroupOffsets, as Apache Kafka’s listShareGroupOffsets does with its ListShareGroupOffsetsHandler.

The call finds the coordinator of every group with one batched FindCoordinator request and sends one DescribeShareGroupOffsets request to each coordinator with every group that it coordinates, each with the partitions of its ListShareGroupOffsetsSpec. COORDINATOR_LOAD_IN_PROGRESS (14) sends a group to the same coordinator again, and COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16) find its coordinator again. A lost connection finds the coordinator again. The call waits with Kafka’s backoff between rounds and stops at default.api.timeout.ms (60 s), where each pending group gets REQUEST_TIMED_OUT (7).

Each group’s offsets hold every answered partition without an error. A partition with no start offset maps to None.

Errors

The call itself does not fail. Each group gets its own KafkaError, such as GROUP_AUTHORIZATION_FAILED (30) or REQUEST_TIMED_OUT (7) at the deadline.

impl AdminClient

pub async fn client_instance_id( &self, timeout: Time, ) -> Result<Option<Uuid>, AdminError>

The client instance id that the broker assigned for metrics push (KIP-714), as Kafka’s KafkaAdminClient.clientInstanceId returns it.

The call waits up to timeout for the first GetTelemetrySubscriptions response. It returns Ok(None) when none has come by then, as Kafka returns null. A zero timeout does not wait.

Errors

Returns AdminError::Transport with ClientError::InvalidArgument for a negative timeout, and with ClientError::TelemetryDisabled when AdminClientConfig::enable_metrics_push is false, as Kafka throws an IllegalArgumentException and an IllegalStateException.

pub async fn close(self)

Close the client: send the terminating telemetry push (KIP-714), if the client pushes metrics and has a subscription, and wait for it, as Kafka’s KafkaAdminClient.close does. The wait is at most about two request timeouts. Dropping the client sends the push in the background instead.

impl AdminClient

pub async fn list_topics( &self, options: ListTopicsOptions, ) -> Result<BTreeMap<String, TopicListing>, AdminError>

Lists the topics of the cluster, as Kafka’s listTopics operation does: one Metadata request for all topics, keeping internal topics only with ListTopicsOptions::list_internal.

Errors

Returns a transport error, and AdminError::Broker with REQUEST_TIMED_OUT (7) at default.api.timeout.ms (60 s).

pub async fn describe_topics( &self, names: &[&str], options: DescribeTopicsOptions, ) -> TopicDescriptions<String>

Describes each topic of names, as Kafka’s describeTopics operation does for a TopicNameCollection.

The client reads the nodes of the cluster with DescribeCluster, then sends DescribeTopicPartitions (KIP-966) and follows its cursor until every topic is complete, at most DescribeTopicsOptions::partition_size_limit_per_response partitions per answer. When the peer does not support DescribeTopicPartitions, such as a controller reached through controller bootstrap, it sends Metadata instead.

The result has one entry for each name:

• The description of the topic.
• INVALID_TOPIC_EXCEPTION (17) for an empty name, which Kafka cannot send.
• The topic error of the answer, such as UNKNOWN_TOPIC_OR_PARTITION (3) or TOPIC_AUTHORIZATION_FAILED (29).
• The error of the call for every topic it did not finish, such as a failed DescribeCluster, a transport error, or REQUEST_TIMED_OUT (7) at default.api.timeout.ms (60 s).

pub async fn describe_topics_by_id( &self, ids: &[Uuid], options: DescribeTopicsOptions, ) -> TopicDescriptions<Uuid>

Describes each topic of ids, as Kafka’s describeTopics operation does for a TopicIdCollection: one Metadata request (v12 or higher, which Kafka needs for topic IDs).

The result has one entry for each ID: the description of the topic, INVALID_TOPIC_EXCEPTION (17) for the nil ID, UNKNOWN_TOPIC_ID (100) for an ID that the answer does not describe, or the error of the call, such as UNSUPPORTED_VERSION (35) for a peer below Metadata v12.

impl AdminClient

pub async fn alter_partition_assignments_with( &mut self, assignments: &BTreeMap<(String, i32), Option<Vec<i32>>>, timeout: Time, options: AlterPartitionReassignmentsOptions, ) -> Result<Vec<PartitionAssignmentOutcome>, AdminError>

Submits exact replica lists with Kafka’s AlterPartitionReassignmentsOptions.

Errors

Returns [krabka_client_core::ClientError::IncompatibleVersion] before sending when allow_replication_factor_change is false and the controller has no v1, as Kafka’s AlterPartitionReassignmentsRequest.Builder refuses to build it. Otherwise a transport, protocol, or top-level broker error.

impl AdminClient

pub async fn force_terminate_transaction( &self, transactional_id: &str, ) -> Result<(), AdminError>

Fences the current producer generation and aborts any ongoing transaction for transactional_id.

This is Kafka’s forceTerminateTransaction operation: it discovers the transaction coordinator and sends InitProducerId with no producer identity and keepPreparedTxn=false. It is safe to call when no transaction is open; the coordinator still advances the producer generation so stale writers are fenced.

The client negotiates InitProducerId v5 or lower, as Kafka’s FenceProducersHandler does. v6 is an unstable version.

The call retries as Apache Kafka’s FenceProducersHandler.handleError does:

• COORDINATOR_LOAD_IN_PROGRESS (14) and CONCURRENT_TRANSACTIONS (51): send the request again to the same coordinator.
• COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16): find the coordinator again, then send the request again.

A FindCoordinator answer of 14 or 15, and a failed or lost connection to the coordinator (including a TLS or SASL handshake that stops before the broker gives a verdict), also make the call find the coordinator again. The call waits between attempts with Kafka’s backoff, and it stops when Kafka’s default default.api.timeout.ms (60 s) elapses. A call whose retries are unresolved at the deadline, or whose attempt is still running, fails with a timeout error, as Kafka gives a TimeoutException.

Errors

Returns AdminError::Protocol for an empty transactional ID, or the coordinator lookup, connection, transport, and broker errors returned by Kafka. Returns ClientError::IncompatibleVersion when the coordinator does not support InitProducerId v5 or lower.

pub async fn fence_producers( &self, transactional_ids: &[&str], ) -> BTreeMap<String, Result<(), AdminError>>

Fences the current producer generation and aborts any ongoing transaction for each ID in transactional_ids, as Kafka’s fenceProducers operation does.

Two transactional IDs can live on two different coordinators, so the client fences them at the same time rather than one at a time. All IDs share one call deadline, as Kafka’s FenceProducersHandler and AdminApiDriver run one invokeDriver for the whole call. The result has one entry for each ID in transactional_ids, in no particular order.

See Self::force_terminate_transaction for the retry behavior and the InitProducerId version that the client negotiates.

Errors

Each entry gets AdminError::Protocol for an empty transactional ID, or the coordinator lookup, connection, transport, and broker errors returned by Kafka for that ID.

pub async fn abort_transaction( &self, spec: &AbortTransactionSpec, ) -> Result<(), AdminError>

Fences the current producer generation and aborts the ongoing transaction that wrote to topic-partition, without contacting the transaction coordinator.

This is Kafka’s abortTransaction operation (KIP-664): it sends WriteTxnMarkers straight to the partition leader with the identity of the transaction to fence, as AbortTransactionHandler does. A consumer that reads describe_producers or the partition’s records supplies spec.

The call retries as Apache Kafka’s AbortTransactionHandler.handleResponse does: NOT_LEADER_OR_FOLLOWER (6) and LEADER_NOT_AVAILABLE (5), from either the leader lookup or the WriteTxnMarkers answer, look the leader up again. A failed or lost connection to the leader does the same. Every other broker error is final. The call waits between attempts with Kafka’s backoff, and it stops when Kafka’s default default.api.timeout.ms (60 s) elapses.

Errors

Returns AdminError::Protocol for an empty topic name, or when the broker’s answer names no result for the requested producer, topic, or partition. Returns AdminError::Broker when the leader lookup or the WriteTxnMarkers call fails, such as with TRANSACTIONAL_ID_AUTHORIZATION_FAILED or FENCED_INSTANCE_ID. Returns AdminError::Transport when the connection to the leader fails.

pub async fn describe_producers( &self, partitions: &[(String, i32)], ) -> BTreeMap<(String, i32), Result<Vec<ProducerStateInfo>, KafkaError>>

Reads the producer state of each requested partition, as Kafka’s describeProducers operation does (KIP-664).

The client finds the leader of each partition with Metadata and sends one DescribeProducers request to each leader, all at the same time, batching every partition of that leader into it, as Kafka’s DescribeProducersHandler with its PartitionLeaderStrategy does. The result has one entry for each requested (topic, partition) pair.

A partition goes back to the leader lookup, and the call finds its leader again after the backoff, when:

• the lookup names no leader yet, such as for an unknown topic or a partition with LEADER_NOT_AVAILABLE;
• the connection to its leader fails or is lost;
• the leader answers NOT_LEADER_OR_FOLLOWER (6) for it.

A slow or failed leader does not delay the other leaders. The call stops at Kafka’s default default.api.timeout.ms (60 s), where each unresolved partition gets REQUEST_TIMED_OUT (7).

Errors

The call itself does not fail. Each partition gets its own KafkaError in the result, such as the final lookup or DescribeProducers error of that partition.

pub async fn list_transactions( &self, filter: &ListTransactionsFilter, ) -> Result<Vec<TransactionListing>, AdminError>

Lists every transaction that the cluster’s coordinators currently hold, as Kafka’s listTransactions operation does (KIP-664).

The client fetches the broker list from Metadata and sends ListTransactions to every broker at the same time, since each transaction coordinator answers only for the transactions it coordinates. filter narrows the state, producer ID, and minimum duration of the listed transactions, as Kafka’s ListTransactionsOptions does; the default matches every transaction.

Errors

Returns AdminError::Broker when any broker gives a top-level error, and AdminError::Transport when the connection to any broker fails. A partial listing from the healthy brokers is not returned; callers that need per-broker results should read AdminError for which broker failed and retry.

pub async fn describe_transaction( &self, transactional_id: &str, ) -> Result<TransactionDescription, AdminError>

Reads the transaction coordinator’s current state for one transactional ID.

This is Kafka’s describeTransactions operation (KIP-664). The client discovers the transaction coordinator for transactional_id and sends DescribeTransactions to that coordinator. An external system calls this to verify the authority of a writer. It joins no group and holds no producer state. The returned producer ID and producer epoch are the only generation from which the coordinator accepts writes.

The call retries as Apache Kafka’s DescribeTransactionsHandler.handleError does:

• COORDINATOR_LOAD_IN_PROGRESS (14): send the request again to the same coordinator.
• COORDINATOR_NOT_AVAILABLE (15) and NOT_COORDINATOR (16): find the coordinator again, then send the request again.

A FindCoordinator answer of 14 or 15, and a failed or lost connection to the coordinator (including a TLS or SASL handshake that stops before the broker gives a verdict), also make the call find the coordinator again. The call waits between attempts with Kafka’s backoff, and it stops when Kafka’s default default.api.timeout.ms (60 s) elapses. A call whose retries are unresolved at the deadline, or whose attempt is still running, fails with a timeout error, as Kafka gives a TimeoutException.

Errors

Returns AdminError::Protocol for an empty transactional ID, and for a response that carries no row for transactional_id. Returns AdminError::Broker when the coordinator lookup fails, or when the coordinator attaches an error code to the row, such as TRANSACTIONAL_ID_NOT_FOUND, TRANSACTIONAL_ID_AUTHORIZATION_FAILED, or NOT_COORDINATOR. Returns AdminError::Transport when the connection to the coordinator fails.

pub async fn describe_transactions( &self, transactional_ids: &[&str], ) -> Result<Vec<TransactionDescription>, AdminError>

Reads the transaction coordinator’s current state for each ID in transactional_ids.

Two transactional IDs can live on two different coordinators, so the client describes them one at a time. Each ID costs one FindCoordinator round trip, one new connection to the coordinator that the lookup names, and one DescribeTransactions round trip. The cost grows linearly with the number of IDs. Callers that describe a large set should expect that cost. The results keep the order of transactional_ids.

Errors

Stops at the first ID that fails and returns its error. Returns AdminError::Protocol for an empty transactional ID, and for a response that carries no row for the ID. Returns AdminError::Broker when the coordinator lookup fails, or when the coordinator attaches an error code to the row. Returns AdminError::Transport when the connection to a coordinator fails.

impl AdminClient

pub async fn describe_user_scram_credentials( &mut self, users: Option<&[String]>, ) -> Result<Vec<UserScramCredentials>, AdminError>

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

impl AdminClient

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

Connects with Kafka’s AdminClientConfig settings. Each call of the client uses the call deadline, retry backoff and retry count of config.

Errors

Returns AdminError::InvalidConfig for invalid settings, and AdminError::Connect with the last cause if no bootstrap address connects.

pub async fn connect_controller_with_config( bootstrap_controllers: &[String], config: AdminClientConfig, ) -> Result<Self, AdminError>

Controller-bootstrap variant of Self::connect_with_config (KIP-919).

Errors

Returns AdminError::InvalidConfig for invalid settings, or a connection, protocol, or broker error when no controller bootstrap endpoint can identify and connect to the active controller.

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

Connects and applies optional client security. None means plaintext, which is identical to AdminClient::connect.

Errors

Returns AdminError::Connect if no bootstrap address accepted the (optionally secured) connection. Its source holds the last cause.

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

Connects with the supplied security and DNS deadline, and keeps the standard admin identity, TCP-connect timeout, and request timeout.

Errors

Returns AdminError::Connect with the last cause if no bootstrap address connects.

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

Connects with the standard plaintext admin policy and a custom DNS deadline.

Errors

Returns AdminError::Connect with the last cause if no bootstrap address connects.

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

Connects with a complete connection-options template.

Errors

Returns AdminError::Connect with the last cause if no bootstrap address accepted the connection.

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

Connects through KIP-919 controller bootstrap addresses rather than broker bootstrap addresses.

Errors

Returns a connection, protocol, or broker error when no controller bootstrap endpoint can identify and connect to the active controller.

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

Controller-bootstrap variant of Self::connect_secured.

Errors

Returns a connection, protocol, or broker error when no controller bootstrap endpoint can identify and connect to the active controller.

pub async fn connect_controller_with_options( bootstrap_controllers: &[String], options: ConnectionOptions, ) -> Result<Self, AdminError>

Controller-bootstrap variant of Self::connect_with_options.

Errors

Returns a connection, protocol, or broker error when no controller bootstrap endpoint can identify and connect to the active controller.

pub async fn connect_with_metadata_recovery( bootstrap_addrs: &[String], options: ConnectionOptions, strategy: MetadataRecoveryStrategy, rebootstrap_trigger: Time, ) -> Result<Self, AdminError>

Connect with explicit KIP-1102 metadata recovery settings.

Errors

Returns an invalid-configuration error for a negative or fractional millisecond trigger, or a connection error when no bootstrap endpoint is reachable.

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

Tries each bootstrap address in order.

Each entry is host:port, which [krabka_client_core::transport::resolve] resolves. The first successful connect wins. Returns AdminError::Connect with the last cause if none responded. The connection is plaintext. See AdminClient::connect_secured.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

Trait Implementations

impl AdminClientLike for AdminClient

fn update_metadata_version<'life0, 'async_trait>( &'life0 mut self, level: i16, upgrade_type: UpgradeType, timeout: Time, ) -> Pin<Box<dyn Future<Output = Result<MetadataVersionUpdate, AdminError>> + Send + 'async_trait>>

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

Finalize metadata.version at level through UpdateFeatures.

fn describe_metadata_quorum<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<MetadataQuorum, AdminError>> + Send + 'async_trait>>

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

Return the live __cluster_metadata partition-zero quorum view.

fn remove_raft_voter<'life0, 'life1, 'async_trait>( &'life0 mut self, cluster_id: Option<&'life1 str>, node_id: i32, directory_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<(), AdminError>> + Send + 'async_trait>>

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

Remove one exact node and directory identity from the metadata quorum, as Kafka’s removeRaftVoter operation does.

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,

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

fn reconcile_topic_replication_factor<'life0, 'life1, 'async_trait>( &'life0 mut self, topic: &'life1 str, replication_factor: i32, timeout: Time, ) -> Pin<Box<dyn Future<Output = Result<TopicReplicationStatus, AdminError>> + Send + 'async_trait>>

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

Converges every partition of topic to replication_factor through AlterPartitionReassignments.

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

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

Returns current replica membership for every selected partition.

fn list_partition_reassignments<'life0, 'life1, 'async_trait>( &'life0 mut self, partitions: &'life1 BTreeMap<String, Vec<i32>>, timeout: Time, ) -> Pin<Box<dyn Future<Output = Result<Vec<PartitionAssignment>, AdminError>> + Send + 'async_trait>>

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

Lists active partition reassignments and their replica membership.

fn alter_partition_assignments<'life0, 'life1, 'async_trait>( &'life0 mut self, assignments: &'life1 BTreeMap<(String, i32), Option<Vec<i32>>>, timeout: Time, ) -> Pin<Box<dyn Future<Output = Result<Vec<PartitionAssignmentOutcome>, AdminError>> + Send + 'async_trait>>

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

Submits exact replica lists; None cancels an active reassignment. A reassignment may change a partition’s replication factor, Kafka’s default; AdminClient::alter_partition_assignments_with can refuse that.

fn unregister_broker<'life0, 'async_trait>( &'life0 mut self, broker_id: i32, ) -> Pin<Box<dyn Future<Output = Result<(), AdminError>> + Send + 'async_trait>>

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

Unregisters one broker from the active KRaft controller.

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

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

Creates topics through the active controller, as Kafka’s KafkaAdminClient.createTopics does.

fn delete_topics<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, names: &'life1 [&'life2 str], options: TopicMutationOptions, ) -> 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,

Deletes topics through the active controller, as Kafka’s KafkaAdminClient.deleteTopics does. The call retries NOT_CONTROLLER (41) and THROTTLING_QUOTA_EXCEEDED (89) as AdminClientLike::create_topics does.

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

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

Adds partitions through the active controller, as Kafka’s KafkaAdminClient.createPartitions does. The call retries NOT_CONTROLLER (41) and THROTTLING_QUOTA_EXCEEDED (89) as AdminClientLike::create_topics does.

fn delete_records<'life0, 'life1, 'async_trait>( &'life0 mut self, ops: &'life1 [DeleteRecordsOp], timeout: Time, ) -> Pin<Box<dyn Future<Output = Result<Vec<DeleteRecordsOutcome>, AdminError>> + Send + 'async_trait>>

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

Deletes records below each requested partition offset.

fn describe_configs<'life0, 'life1, 'async_trait>( &'life0 mut self, resources: &'life1 [ConfigResource], options: DescribeConfigsOptions, ) -> Pin<Box<dyn Future<Output = Result<DescribeConfigsResults, AdminError>> + Send + 'async_trait>>

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

Describes the configs of each resource of resources, as Kafka’s describeConfigs operation does.

fn incremental_alter_configs<'life0, 'life1, 'async_trait>( &'life0 mut self, configs: &'life1 BTreeMap<ConfigResource, Vec<AlterConfigOp>>, options: IncrementalAlterConfigsOptions, ) -> Pin<Box<dyn Future<Output = Result<AlterConfigsResults, AdminError>> + Send + 'async_trait>>

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

Changes the configs of each resource of configs, as Kafka’s incrementalAlterConfigs operation does. With IncrementalAlterConfigsOptions::validate_only the broker validates the changes and applies none.

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,

Upserts SCRAM-SHA-512 credentials, deletes them, or both, in a single call.

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.

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,

Lists ACLs that match filter.

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,

Creates the supplied ACLs.

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,

Deletes every ACL that matches any of filters.

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,

Reads the quotas of the user entity {user: username}.

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,

Applies ops to the user entity {user: username}. Returns the error the broker gave for the entity, or None on success. No request is sent for an empty ops.

fn create_delegation_token<'life0, 'life1, 'async_trait>( &'life0 mut self, options: &'life1 CreateDelegationTokenOptions, ) -> Pin<Box<dyn Future<Output = Result<DelegationToken, AdminError>> + Send + 'async_trait>>

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

Creates a delegation token, as Kafka’s createDelegationToken operation does.

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

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

Renews the delegation token with hmac, as Kafka’s renewDelegationToken operation does, and returns its new expiry in epoch milliseconds.

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

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

Expires the delegation token with hmac, as Kafka’s expireDelegationToken operation does, and returns the expiry that the broker set, in epoch milliseconds. A token that expires now gets the broker’s clock at the time of the request.

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

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

Describes the delegation tokens that options selects, as Kafka’s describeDelegationToken operation does.