Struct KraftController 

pub struct KraftController { /* private fields */ }

Cheap, cloneable handle to the running engine: holds the command sender and the watch receivers the broker/handle read.

Implementations

impl KraftController

pub fn node_id(&self) -> NodeId

The node id this controller runs as.

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

A snapshot of the latest applied MetadataImage.

pub fn finalized_features_epoch(&self) -> i64

The finalized-features epoch of the published MetadataImage: the offset of the last record of the newest batch or snapshot that carried a FeatureLevelRecord, or -1 before the first. Kafka’s controller listener answers ApiVersions with it, from FeaturesPublisher.

The engine publishes it after the image, so a caller that reads it before Self::current_image never gets an epoch ahead of the features.

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

Watch the published MetadataImage.

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

Watch the fault that stopped the engine: None while it runs, and the reason once it stops over a fault Kafka halts the process over. That is the I/O error of the first write to the metadata log directory that fails (KIP-858), or a controller activation that failed.

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

Watch the current leader id.

pub fn quorum_snapshot(&self) -> QuorumStateSnapshot

A synchronous snapshot of consensus state (the handle’s quorum_state() reads this without an mpsc round-trip — the engine republishes it on every event). Cheap watch borrow + clone.

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

Submit a metadata change. On the leader, appends the batch at the current leader epoch and returns once it is committed (HWM ≥ the appended end offset) AND applied, surfacing the first per-record rejection. On a follower, returns RaftError::NotLeader with the leader hint; the handle layer forwards via forward_submit_to.

Errors

• RaftError::Metadata if a record fails validate.
• RaftError::NotLeader if this node is not the leader.
• RaftError::Shutdown if the engine task is gone.

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 on the leader. It can be installed once.

Errors

Returns an error if a hook is already installed.

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

Submit a leader-local mutation only while its admission image is current. Pending metadata must commit before a caller can plan a guarded mutation.

Errors

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

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

Submit generation-bound delegation-token mutations.

Errors

Returns a leadership, shutdown, or mutation-rejection error.

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

Atomically finalize kraft.version in the Raft control log.

Errors

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

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

Run the leader-side checks of Self::finalize_kraft_version and append nothing.

Errors

Returns the error finalize_kraft_version would return for the same request.

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

A structured snapshot of consensus state for the broker’s DescribeQuorum admin view.

Errors

Returns RaftError::Shutdown if the engine task is gone.

pub async fn metadata_fetch( &self, fetch_offset: i64, max_size: ByteSize, replica: Option<ReplicaKey>, ) -> Result<MetadataFetchSlice, RaftError>

Read a committed __cluster_metadata slice for an observer’s API_KEY_METADATA_FETCH (1004). replica names the observer that asked, when it is a replica, so a leader can track its progress.

Errors

Returns RaftError::Shutdown if the engine task is gone.

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

Serialize the current image to a KIP-630 checkpoint under the data dir.

Errors

Returns RaftError if serialization or the file write fails.

pub async fn inject_event(&self, event: Event) -> Result<(), RaftError>

Inject a raw core Event into the loop (test/driver entrypoint and the internal feedback path for peer-RPC responses).

Errors

Returns RaftError::Shutdown if the engine task is gone.

pub async fn deliver(&self, inbound: Inbound) -> Result<(), RaftError>

Deliver an inbound peer RPC to the engine.

Errors

Returns RaftError::Shutdown if the engine task is gone.

pub async fn shutdown(&self)

Stop the engine task gracefully and wait for its log to close.

A leader resigns first and sends EndQuorumEpoch, and the engine keeps taking part in the quorum until a successor is known or its graceful shutdown window runs out, as Kafka’s KafkaRaftClient.shutdown does. A follower or an observer stops at once.

pub async fn halt(&self)

Stop the engine task at once, handing nothing over, and wait for its log to close.

This is the stop of a halted process: a fatal fault, or a test that models a crash. The quorum finds out through its fetch watchdog.

impl KraftController

pub fn spawn( config: KraftConfig, log: KraftLog, data_dir: PathBuf, ) -> Result<Self, RaftError>

Build the engine over an already-opened KraftLog and spawn its loop task. Recovery (snapshot + replay + quorum-state file) is wired by Self::open; this lower-level entrypoint takes the seed state directly and is used by tests/drivers that supply their own KraftLog.

The returned handle’s loop runs until Self::shutdown (or all handles drop). data_dir is where the engine writes the quorum-state file and checkpoints.

Errors

Returns an error if starting the engine fails. A fresh controller has no recovered downgrade boundary, so it needs no downgrade-checkpoint recovery.

pub fn open( data_dir: PathBuf, me: NodeId, cluster_id: Uuid, directory_id: Uuid, bootstrap_voters: VoterSet, timeouts: QuorumTimeouts, heartbeat_interval: Option<Time>, controller_fetch_miss_limit: ControllerFetchMissLimit, metadata_raft_command_queue_capacity: MetadataRaftCommandQueueCapacity, metadata_raft_fetch_max: MetadataRaftFetchMax, peers: Arc<dyn PeerSender>, snapshot_interval_records: u64, max_bytes_between_snapshots: ByteSize, max_snapshot_interval: Time, metadata_snapshot_fetch_max: MetadataSnapshotFetchMax, metadata_log: MetadataLogConfig, activation: Activation, event_queue_recorder: Arc<dyn EventQueueRecorder>, validate_image: impl Fn(&MetadataImage) -> Result<(), RaftError>, ) -> Result<Self, RaftError>

Open the engine over data_dir: recover the MetadataImage from the latest checkpoint + replay committed log batches, and seed the durable QuorumState from the node-local quorum-state file. The bootstrap voter set/cluster id is used only when no quorum-state file exists yet.

The bootstrap checkpoint, at offset 0 and epoch 0, is not a snapshot of the log. The engine takes its KIP-853 control state, but it does not apply its metadata records to the image. Those records replace the bootstrap records of activation when there are any, and the leader writes them to an empty log, as Kafka’s QuorumController.handleLoadBootstrap keeps them for its activation.

Errors

Returns RaftError if the log/checkpoint cannot be opened or read, or validate_image refuses the committed image, or that image with every feature level the log holds above its high watermark, before checkpoint repair or an engine task can start.

Trait Implementations

impl Clone for KraftController

fn clone(&self) -> KraftController

Returns a duplicate of the value.

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

Performs copy-assignment from source.