Struct ControllerHandle 

pub struct ControllerHandle { /* private fields */ }

Handle returned by Controller::start.

Owns the live KraftController engine and the listener task. Drop is NOT a clean shutdown — call Self::shutdown (or Self::cancel) to drain the listener + stop the engine before the runtime is torn down.

Implementations

impl ControllerHandle

pub fn read_snapshot_range( &self, position: i64, max_bytes: i32, ) -> SnapshotRange

Read up to max_bytes of the latest metadata snapshot starting at position. Reads the engine’s .checkpoint artifacts directly (the engine writes a bare KIP-630 checkpoint, no .meta sidecar).

position and max_bytes stay the raw KIP-595 FetchSnapshot int64 and int32: both are byte offsets into an on-disk checkpoint that the broker’s handler forwards straight off the wire, so there is no domain layer between the decode and the slice index for a quantity to occupy.

pub async fn fetch_snapshot( &self, version: i16, body: Bytes, ) -> Result<Bytes, RaftError>

Answer one KIP-630 FetchSnapshot request through the engine, as the controller listener answers it.

body is the request body at version, and the answer is the response body at that version. The engine follows Kafka’s KafkaRaftClient.handleFetchSnapshotRequest: it serves exactly the snapshot the request names, and refuses another topic or partition, a stale or unknown CurrentLeaderEpoch, another cluster id, and a position outside the snapshot. It also fills CurrentLeader and the leader’s endpoint. A broker listener that answered from the checkpoint directly would answer the same request differently.

Errors

Returns RaftError::Shutdown if the engine task is gone, or if it dropped the request because the body does not decode.

pub async fn trigger_snapshot(&self) -> Result<(), RaftError>

Manually trigger a metadata snapshot (KIP-630 checkpoint) on this node.

Errors

Returns RaftError if serialization or the file write fails.

impl ControllerHandle

pub async fn change_membership( &self, new_voters: BTreeSet<NodeId>, ) -> Result<(), RaftError>

Reconcile a single-node voter-set delta through the KIP-853 engine.

Errors

Rejects multi-node batch changes; KIP-642 is a separate operation.

pub fn add_learner( &self, node_id: NodeId, node: Node, ) -> Ready<Result<(), RaftError>>

Stage a caught-up observer identity for later voter promotion.

Errors

The observer catches up by fetching from the leader; this call does not alter quorum membership.

pub async fn add_voter( &self, req: AddVoter, ) -> Result<ReconfigOutcome, RaftError>

Add a caught-up controller voter through the Raft control log.

Errors

Returns a validation, leadership, timeout, or storage error from Raft.

pub async fn check_add_voter( &self, req: AddVoter, ) -> Result<ReconfigOutcome, RaftError>

Run the leader-local admission checks of AddVoterHandler for req without contacting the candidate or appending anything.

Errors

Returns the refusal the same request would get from Self::add_voter for a pending change, an unset high watermark, kraft.version below 1, an uncommitted voters record or a duplicate voter id.

pub async fn remove_voter( &self, req: RemoveVoter, ) -> Result<ReconfigOutcome, RaftError>

Remove the exact node/directory pair through the Raft control log.

Errors

Returns a validation, leadership, timeout, or storage error from Raft.

pub async fn update_voter( &self, req: UpdateVoter, ) -> Result<ReconfigOutcome, RaftError>

Update the exact voter’s endpoint and supported feature range.

Errors

Returns a validation, leadership, timeout, or storage error from Raft.

pub async fn finalize_kraft_version( &self, version: u16, ) -> Result<ReconfigOutcome, RaftError>

Atomically append KRaftVersionRecord and the initial VotersRecord.

Errors

Returns an unsupported-version, leadership, timeout, or storage error.

pub async fn validate_kraft_version( &self, version: u16, ) -> Result<ReconfigOutcome, RaftError>

Run the checks of Self::finalize_kraft_version and write nothing.

Errors

Returns the error finalize_kraft_version would return for the same request.

impl ControllerHandle

pub async fn metadata_records( &self, fetch_offset: u64, max_size: ByteSize, ) -> MetadataFetchSlice

Read committed __cluster_metadata entries starting at fetch_offset, encoded as Kafka record batches for an observer.

pub async fn fetch_metadata_from( &self, addr: SocketAddr, fetch_offset: u64, max_size: ByteSize, ) -> Result<KrabkaMetadataFetchResponse, RaftError>

Dial a controller-listener addr and issue one API_KEY_METADATA_FETCH. Used by broker-only observers to pull committed __cluster_metadata.

Errors

• RaftError::Network if the dial or request fails.
• RaftError::Protocol if the response cannot be decoded.

impl ControllerHandle

pub async fn submit_change_if_image( &self, records: Vec<MetadataRecord>, expected_image: Arc<MetadataImage>, ) -> Result<SubmitChangeResult, RaftError>

Submit a mutation admitted against this controller’s published image. The engine checks that image before appending; this method never forwards a mutation whose admission happened on a different controller.

Errors

Returns a leadership, stale-image, uncommitted-tail, or submission error.

pub async fn submit_change( &self, records: Vec<MetadataRecord>, ) -> Result<SubmitChangeResult, RaftError>

Submit a batch of metadata records. Returns Ok(()) once committed AND applied on the leader. Pre-validation lives in the engine. On a follower (NotLeader with a known leader), forwards directly to the leader’s controller listener via API_KEY_SUBMIT_CHANGE.

Errors

Returns an error if validation, replication, or forwarding fails.

pub async fn submit_delegation_token_mutations( &self, mutations: Vec<DelegationTokenMutation>, ) -> Result<SubmitChangeResult, RaftError>

Submit token mutations whose expected records are checked by the leader in the same engine turn as the append.

Errors

Returns a leadership, transport, protocol, or mutation-rejection error.

pub async fn forward_raw( &self, api_key: i16, version: i16, body: Bytes, ) -> Result<Bytes, RaftError>

Forward a raw Kafka-wire request for api_key/version to the current quorum leader’s controller listener and return its raw response bytes, unmodified.

Used by the broker’s DescribeQuorum handler: Kafka’s KafkaApis forwards DescribeQuorum from the broker listener to the active controller unconditionally (forwardToController), so a combined or controller node that is not itself the leader must reach the leader the same way a broker-only observer already does.

Errors

RaftError::NotLeader when no leader is known or the leader’s address cannot be resolved, RaftError::Network when the dial or round trip fails.

impl ControllerHandle

pub fn current_image(&self) -> Arc<MetadataImage>

Current metadata snapshot (cheap; Arc clone).

pub fn install_registration_admission( &self, admission: Arc<dyn RegistrationAdmission>, ) -> Result<(), &'static str>

Install the hook every batch that registers a broker anew passes through while this node leads. See RegistrationAdmission.

Errors

Returns an error if a hook is already installed.

pub fn controller_bound_addr(&self) -> SocketAddr

The address the controller listener actually bound to.

pub fn watch_leader(&self) -> Receiver<Option<NodeId>>

Subscribe to leader-id changes.

pub fn watch_fatal(&self) -> Receiver<Option<String>>

Subscribe to the controller’s fatal fault: the message of the fault that made it stop itself, and None until then.

Kafka handles such a fault (a FeatureControlManager replay of a feature level that the controller does not support) with a ProcessTerminatingFaultHandler, which halts the process. A library cannot halt its host, so the controller stops itself, publishes the message here before it does, and leaves the process to its embedder: the broker turns it into its own shutdown. The channel closes without a value when the controller is shut down for any other reason.

pub fn watch_image(&self) -> Receiver<Arc<MetadataImage>>

Subscribe to metadata-image changes.

pub fn quorum_state(&self) -> QuorumState

Snapshot the controller’s current quorum state. Used by the broker’s DescribeQuorum (api_key=55, KIP-595) handler. Cheap — a watch borrow of the engine’s published snapshot.

pub fn quorum_snapshot(&self) -> QuorumStateSnapshot

Snapshot the full consensus state.

pub fn quorum_committed_offset(&self) -> i64

Highest __cluster_metadata offset the quorum has committed, as this node last heard it, or -1 before the first committed record.

This is the offset a readiness probe compares against. It differs from QuorumState::last_applied_index on a follower that is still replaying: last_applied_index is clamped to what this node has applied, while this is what the leader said it had committed.

pub fn voted_directory_id(&self) -> Option<Uuid>

Directory identity voted for in the current leader epoch, if any.

pub async fn shutdown(self)

Drain the listener and stop the engine. Idempotent in practice.

pub async fn cancel(&self)

Stop the engine and cancel the controller listener without consuming self. Used by BrokerHandle::shutdown where the controller is behind an Arc. Idempotent. Awaits the listener task so the OS port is released before returning.

The engine stops gracefully, before the listener does: a leader resigns, and its successor’s BeginQuorumEpoch still reaches the listener, which is how the engine learns that the handover is complete.

pub async fn halt(&self)

Stop the engine at once, handing nothing over, and cancel the listener, as a halted process stops. The quorum finds out through its fetch watchdog. Idempotent.