krabka_raft/kraft/controller/
activation.rs

1//! Controller activation: the records that a new leader writes to a metadata
2//! log that holds no `metadata.version`.
3//!
4//! Kafka's active controller writes them in its `CompleteActivationEvent`,
5//! with `ActivationRecordsGenerator.recordsForEmptyLog`. They are the
6//! bootstrap records, from the bootstrap checkpoint of a dynamic format or
7//! else from the `bootstrap.checkpoint` of a static one, and the cluster-level
8//! `min.insync.replicas` when the bootstrap records enable ELR. A broker never
9//! writes them. Every replica, a broker-only observer too, applies them from
10//! the log. A controller that cannot write them stops over a fatal fault, as
11//! Kafka's `fatalFaultHandler` halts the process.
12
13use {
14    super::Engine,
15    crate::error::RaftError,
16    krabka_metadata::{
17        BrokerConfigRecord, DEFAULT_BROKER_CONFIG_NODE_ID, MetadataRecord,
18        metadata_version::{ELR_VERSION_FEATURE, METADATA_VERSION_FEATURE},
19    },
20    tokio::sync::oneshot,
21};
22
23/// Kafka's `TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG`.
24const MIN_IN_SYNC_REPLICAS_CONFIG: &str = "min.insync.replicas";
25
26/// The `failureMessage` Kafka's `QuorumController` gives its fatal fault
27/// handler when the activation records cannot be written.
28const ACTIVATION_FAULT: &str = "exception while completing controller activation";
29
30/// What a new leader of an empty metadata log writes: the inputs of Kafka's
31/// `ActivationRecordsGenerator.recordsForEmptyLog`.
32#[derive(Debug, Clone, PartialEq, krabka_macros::FieldDefaults)]
33pub struct Activation {
34    /// The bootstrap records. An empty list writes nothing.
35    pub bootstrap_records: Vec<MetadataRecord>,
36    /// This node's static `min.insync.replicas`: Kafka's
37    /// `ConfigurationControlManager.getStaticallyConfiguredMinInsyncReplicas`.
38    /// It becomes the cluster-level `min.insync.replicas` when the bootstrap
39    /// records enable ELR. Default: `1`, Kafka's default.
40    #[default(1)]
41    pub default_min_insync_replicas: i32,
42}
43
44/// The level of the last `FeatureLevelRecord` for `name` in `records`, or
45/// `None` when they hold none. The last record wins, as it does in Kafka's
46/// `BootstrapMetadata` and on replay.
47fn bootstrap_feature_level(records: &[MetadataRecord], name: &str) -> Option<i16> {
48    records.iter().rev().find_map(|record| match record {
49        MetadataRecord::V1FeatureLevel(feature) if feature.name == name => Some(feature.level),
50        _ => None,
51    })
52}
53
54/// The `metadata.version` level of the last `FeatureLevelRecord` for it in
55/// `records`, or `None` when they hold none.
56pub(super) fn bootstrap_metadata_version(records: &[MetadataRecord]) -> Option<i16> {
57    bootstrap_feature_level(records, METADATA_VERSION_FEATURE)
58}
59
60/// The records of the activation of an empty log, in Kafka's order: the
61/// bootstrap records, and then the cluster-level `min.insync.replicas` at the
62/// static value when the bootstrap records finalize
63/// `eligible.leader.replicas.version` above 0. They go to the log as one
64/// batch, which commits whole, as Kafka writes them in one atomic batch or
65/// metadata transaction.
66pub(super) fn activation_records(activation: &Activation) -> Vec<MetadataRecord> {
67    let mut records = activation.bootstrap_records.clone();
68    if bootstrap_feature_level(&records, ELR_VERSION_FEATURE).is_some_and(|level| level > 0) {
69        records.push(MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
70            node_id: DEFAULT_BROKER_CONFIG_NODE_ID,
71            config_name: MIN_IN_SYNC_REPLICAS_CONFIG.to_owned(),
72            config_value: Some(activation.default_min_insync_replicas.to_string()),
73        }));
74    }
75    records
76}
77
78/// Refuse bootstrap metadata that holds records but does not finalize
79/// `metadata.version`, as Kafka's `BootstrapMetadata` does.
80///
81/// The leader writes the bootstrap records to a log that holds no
82/// `metadata.version`. Records that do not finalize one would leave the log
83/// in that state, and the next leader would write them again.
84///
85/// # Errors
86///
87/// Returns [`RaftError::Startup`] when `records` is not empty and holds no
88/// `FeatureLevelRecord` for `metadata.version`, with the message of Kafka's
89/// `BootstrapMetadata.fromRecords`. It also returns it when the last such
90/// record sets level 0, which removes the feature and which no
91/// `MetadataVersion` has.
92pub(super) fn check_bootstrap_records(
93    records: &[MetadataRecord],
94    source: &str,
95) -> Result<(), RaftError> {
96    if records.is_empty() {
97        return Ok(());
98    }
99    match bootstrap_metadata_version(records) {
100        Some(level) if level > 0 => Ok(()),
101        Some(level) => Err(RaftError::Startup(format!(
102            "No MetadataVersion with feature level {level} in the bootstrap metadata from \
103             {source}"
104        ))),
105        None => Err(RaftError::Startup(format!(
106            "No FeatureLevelRecord for {METADATA_VERSION_FEATURE} was found in the bootstrap \
107             metadata from {source}"
108        ))),
109    }
110}
111
112impl Engine {
113    /// Write the activation records when the log of this new leader holds no
114    /// `metadata.version`: Kafka's `CompleteActivationEvent`.
115    ///
116    /// The leader calls this directly after it appends the `LeaderChange`
117    /// batch of its epoch. So the records come before every other write of
118    /// the epoch, as Kafka's activation event is the first event of the
119    /// active controller. The log holds `metadata.version` when the committed
120    /// image finalizes it. It also holds it when a batch that an earlier
121    /// leader appended and did not commit finalizes it, because that batch
122    /// commits with this epoch. Kafka makes the same decision after its
123    /// controller has replayed the log up to the epoch start.
124    ///
125    /// The records go through [`Self::on_activation_change`], so the leader
126    /// validates and encodes them as it does every other write. Nothing waits
127    /// for them to commit. A leader that refuses them, or that cannot read
128    /// its own log tail to decide, sets [`Engine::activation_fault`]. The
129    /// engine then stops, and the controller stops over the fault, as Kafka's
130    /// `QuorumController` gives the failure to its `fatalFaultHandler`.
131    pub fn complete_activation(&mut self) {
132        let Some(level) = bootstrap_metadata_version(&self.activation.bootstrap_records)
133            .filter(|level| *level > 0)
134        else {
135            return;
136        };
137        match self.replay_earlier_epoch_tail(|_, _| {}) {
138            Ok(image) if image.finalized_metadata_version().is_some() => return,
139            Ok(_) => {}
140            Err(error) => {
141                self.activation_fault = Some(format!("{ACTIVATION_FAULT}: {error}"));
142                return;
143            }
144        }
145        let records = activation_records(&self.activation);
146        tracing::info!(
147            records = records.len(),
148            metadata_version = level,
149            "kraft: performing controller activation. The metadata log appears to be empty: \
150             appending the bootstrap records"
151        );
152        let (reply, mut outcome) = oneshot::channel();
153        self.on_activation_change(&records, reply);
154        if let Ok(Err(error)) = outcome.try_recv() {
155            self.activation_fault = Some(format!("{ACTIVATION_FAULT}: {error}"));
156        }
157    }
158}
krabka_raft/kraft/controller/activation.rs

krabka_raft/kraft/controller/
activation.rs
