krabka_metadata/
records.rs

1//! Versioned metadata records.
2//!
3//! wincode encodes a [`MetadataRecord`] as its variant index, then the fields
4//! of the variant in declaration order. Nothing is length-prefixed, so a reader
5//! cannot skip a variant it does not know: it fails to decode it. The variant
6//! order and every field layout are part of the krabka 1.x on-disk contract. A
7//! new variant goes at the end. The golden bytes in `wincode_contract.rs` pin
8//! one value of every variant.
9
10use std::collections::BTreeMap;
11
12pub use krabka_ids::LeaderEpoch;
13pub use krabka_voters::NodeId;
14use serde::{Deserialize, Serialize};
15use uuid::Uuid;
16
17use crate::{
18    break_glass::BreakGlassProposalRecord, voters::VoterSet, write_freeze::TopicFreezeRecord,
19};
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct TopicRecord {
23    pub name: String,
24    pub topic_id: Uuid,
25    pub partitions: i32,
26    pub replication_factor: i16,
27}
28
29fn default_partition_epoch() -> i32 {
30    -1
31}
32
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
34pub struct PartitionRecord {
35    pub topic: String,
36    pub partition: i32,
37    pub leader: NodeId,
38    pub replicas: Vec<NodeId>,
39    pub isr: Vec<NodeId>,
40    /// Per-partition leader epoch (KIP-320). Bumped on every leader change.
41    /// Older on-disk metadata is not migrated. `#[serde(transparent)]` on
42    /// [`LeaderEpoch`] keeps the on-disk bincode bytes identical to a bare
43    /// `i32`.
44    pub leader_epoch: LeaderEpoch,
45    /// Replicas being added in an in-flight reassignment. Empty when no
46    /// reassignment in flight. KIP-455.
47    pub adding_replicas: Vec<NodeId>,
48    /// Replicas being removed in an in-flight reassignment. Empty when
49    /// no reassignment in flight. KIP-455.
50    pub removing_replicas: Vec<NodeId>,
51    /// KIP-858: the log-directory UUID that hosts each replica, parallel to
52    /// [`Self::replicas`] in the same index order. The reserved ids are
53    /// Kafka's `DirectoryId` sentinels (see [`crate::directory_id`]):
54    /// [`MIGRATING`](crate::directory_id::MIGRATING), the nil UUID, is a
55    /// replica placed before `3.7-IV2` or on a broker with no log-dir ids,
56    /// and [`UNASSIGNED`](crate::directory_id::UNASSIGNED) is a replica the
57    /// controller placed on a multi-dir broker that has not yet reported its
58    /// `AssignReplicasToDirs`. An empty list reads as every replica
59    /// `MIGRATING`, as Kafka's `PartitionRegistration` reads it. The
60    /// controller matches this against the replica slot of a broker to map
61    /// the failed-dir UUID of that broker to the partitions it must fail
62    /// over.
63    pub directories: Vec<Uuid>,
64    /// KIP-631: per-partition state epoch. It increments on every state
65    /// change, such as a leader election, an ISR change, or a reassignment.
66    /// It is 0 on creation. The default of -1 matches the KIP-631 schema
67    /// default, for compatibility with records written before this field
68    /// existed.
69    #[serde(default = "default_partition_epoch")]
70    pub partition_epoch: i32,
71}
72
73/// KIP-858 directory-assignment delta.
74///
75/// A broker reports which log-dir UUID hosts its replica of `(topic,
76/// partition)`. Apply treats it as a DELTA: it sets ONLY the slot of the
77/// reporting replica in `PartitionRecord.directories` and never touches leader,
78/// isr, replicas, adding, or removing. It therefore cannot clobber a concurrent
79/// reassignment or ISR change. On the `KRaft` log it is encoded as Kafka's
80/// standard `PartitionChangeRecord` with only the directories field set.
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82pub struct PartitionDirAssignmentRecord {
83    pub topic: String,
84    pub partition: i32,
85    /// The reporting broker (must be a replica of the partition).
86    pub replica: NodeId,
87    /// The log-directory UUID hosting this broker's replica.
88    pub directory: Uuid,
89}
90
91/// Diskless offset-sequencer delta: advance a partition's committed
92/// next-offset by `count`.
93///
94/// Applied as a delta, never a full-record replace, so sequential advances on
95/// the committed metadata log yield a gap-free, strictly-monotonic, unique
96/// offset sequence. On the `KRaft` log it still uses a Krabka-private carrier.
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub struct PartitionOffsetAdvanceRecord {
99    pub topic: String,
100    pub partition: i32,
101    /// Offsets consumed by the produce group. The producer's base offset is the
102    /// committed next-offset before this increment applies.
103    pub count: i64,
104}
105
106/// KIP-966 eligible-leader state for one partition. Empty vectors clear it.
107#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
108pub struct PartitionElrRecord {
109    pub topic: String,
110    pub partition: i32,
111    pub eligible_leader_replicas: Vec<NodeId>,
112    pub last_known_elr: Vec<NodeId>,
113}
114
115/// KIP-704 state of an uncleanly elected leader.
116#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
117#[repr(i8)]
118pub enum LeaderRecoveryState {
119    #[default]
120    Recovered = 0,
121    Recovering = 1,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125pub struct PartitionRecoveryRecord {
126    pub topic: String,
127    pub partition: i32,
128    pub state: LeaderRecoveryState,
129}
130
131/// A Kafka `PartitionChangeRecord` can carry assignment, ELR, and recovery
132/// changes together. This record preserves that atomic update on replay.
133#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
134pub struct PartitionUpdateRecord {
135    pub partition: PartitionRecord,
136    pub eligible_leader_replicas: Option<Vec<NodeId>>,
137    pub last_known_elr: Option<Vec<NodeId>>,
138    pub recovery_state: Option<LeaderRecoveryState>,
139}
140
141/// A single named listener endpoint advertised by a broker.
142///
143/// It is stored as a list on [`BrokerRegistrationRecord::endpoints`], so
144/// KRaft-style metadata can advertise per-listener `host:port` and protocol
145/// triples to clients on `Metadata` v9+. A legacy single-listener broker leaves
146/// the list empty and relies on the top-level `host` and `port` fields.
147#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
148pub struct BrokerEndpoint {
149    /// Listener name (e.g. `"PLAINTEXT"`, `"SSL"`, `"SASL_SSL"`).
150    pub name: String,
151    pub host: String,
152    pub port: u16,
153    pub protocol: krabka_security::ListenerProtocol,
154}
155
156#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
157pub struct BrokerRegistrationRecord {
158    pub node_id: NodeId,
159    /// KIP-903 broker epoch: the raft log offset at which this registration
160    /// record committed. The controller leader assigns it at append time in
161    /// `on_submit_change`. A freshly-built literal carries `0` until the
162    /// leader stamps it. `AlterPartition` uses it to fence stale replicas
163    /// from the ISR.
164    pub broker_epoch: i64,
165    /// KIP-631: UUID that identifies this specific process invocation of the
166    /// broker. Generated once at first boot and persisted in
167    /// `{log_dir}/incarnation_id`. A JVM controller uses it to detect
168    /// broker restarts and fence stale replica memberships.
169    #[serde(default)]
170    pub incarnation_id: uuid::Uuid,
171    /// Legacy single-listener host, used as inter-broker default and by
172    /// pre-v9 `Metadata` responses. v9+ projects [`Self::endpoints`].
173    pub host: String,
174    pub port: u16,
175    pub rack: Option<String>,
176    /// Per-listener endpoints. Empty on records written before this
177    /// field was added; populated from
178    /// `BrokerConfig::effective_listeners()` for self-registration.
179    pub endpoints: Vec<BrokerEndpoint>,
180    /// KIP-858 stable IDs for the broker's online log directories.
181    /// Empty at metadata versions before `3.7-IV2` and in legacy snapshots.
182    #[serde(default)]
183    pub log_dirs: Vec<uuid::Uuid>,
184    /// KIP-631 fencing state. Kafka writes a new registration fenced (the
185    /// `RegisterBrokerRecord` schema default) and unfences it with a
186    /// [`BrokerRegistrationChangeRecord`] once the broker catches up.
187    pub fenced: bool,
188    /// KIP-841: the broker is in controlled shutdown. Only a
189    /// [`BrokerRegistrationChangeRecord`] sets it; only a new registration
190    /// clears it.
191    pub in_controlled_shutdown: bool,
192    /// KIP-1066 cordoned log directories. `None` until the broker first
193    /// reports them in a heartbeat, as in Kafka's `BrokerRegistration`. Only
194    /// carried from `metadata.version` `4.3-IV0`.
195    pub cordoned_log_dirs: Option<Vec<uuid::Uuid>>,
196    /// KIP-584 feature ranges advertised by this broker at registration.
197    /// Empty only for legacy Krabka snapshots written before the ranges were
198    /// retained in the image.
199    #[serde(default)]
200    pub features: BTreeMap<String, (i16, i16)>,
201}
202
203/// KIP-919 controller registration persisted in the metadata log.
204///
205/// Static controller voters still register their current process incarnation,
206/// listener endpoints, and supported feature ranges.  Keeping that state in
207/// the image lets every controller replay the same registration and gives JVM
208/// peers the `RegisterControllerRecord` they expect in a mixed quorum.
209#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
210pub struct ControllerRegistrationRecord {
211    pub node_id: NodeId,
212    pub incarnation_id: Uuid,
213    pub zk_migration_ready: bool,
214    pub endpoints: Vec<BrokerEndpoint>,
215    pub features: BTreeMap<String, (i16, i16)>,
216}
217
218#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
219pub struct DeleteTopicRecord {
220    pub name: String,
221}
222
223/// KIP-919 / `UnregisterBroker` (`api_key` 64), Kafka's
224/// `UnregisterBrokerRecord` (metadata apiKey 1).
225///
226/// Marks a broker as permanently unregistered: the admin operator confirms that
227/// the broker is gone for good and asks the cluster to drop its registration
228/// entry from the metadata image. Later `Metadata` responses no longer
229/// advertise the endpoints of the broker, and clients stop routing to it.
230///
231/// As in `ClusterControlManager.replay(UnregisterBrokerRecord)`, the record
232/// removes the registration only when `broker_epoch` is the epoch of the
233/// current registration. An apply against an unknown `node_id` or another
234/// epoch does nothing, so a second apply is a no-op.
235#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
236pub struct UnregisterBrokerRecord {
237    pub node_id: NodeId,
238    /// The epoch of the registration this record removes.
239    pub broker_epoch: i64,
240}
241
242/// A change to the fencing state of a broker registration, as Kafka's
243/// `BrokerRegistrationFencingChange` encodes it in
244/// `BrokerRegistrationChangeRecord.Fenced`.
245#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
246pub enum FencingChange {
247    /// `-1`: the broker has been unfenced.
248    Unfence,
249    /// `0`: no change.
250    None,
251    /// `1`: the broker has been fenced.
252    Fence,
253}
254
255impl FencingChange {
256    /// The `Fenced` byte Kafka writes for this change.
257    #[must_use]
258    pub fn wire_value(self) -> i8 {
259        match self {
260            Self::Unfence => -1,
261            Self::None => 0,
262            Self::Fence => 1,
263        }
264    }
265
266    /// Reads a `Fenced` byte, or `None` for a value Kafka does not define.
267    #[must_use]
268    pub fn from_wire(value: i8) -> Option<Self> {
269        match value {
270            -1 => Some(Self::Unfence),
271            0 => Some(Self::None),
272            1 => Some(Self::Fence),
273            _ => None,
274        }
275    }
276
277    /// The fenced state after this change applies to `fenced`.
278    #[must_use]
279    pub fn apply(self, fenced: bool) -> bool {
280        match self {
281            Self::Unfence => false,
282            Self::None => fenced,
283            Self::Fence => true,
284        }
285    }
286}
287
288/// Kafka's `BrokerRegistrationChangeRecord` (metadata apiKey 17): a delta on
289/// the registration of one broker.
290///
291/// As in `ClusterControlManager.replay(BrokerRegistrationChangeRecord)`, the
292/// image applies it only when `broker_epoch` is the epoch of the current
293/// registration; against an unknown broker or another epoch it does nothing.
294/// `in_controlled_shutdown` true is Kafka's `IN_CONTROLLED_SHUTDOWN` (`1`) and
295/// false is `NONE` (`0`); Kafka has no change that leaves controlled shutdown,
296/// only a new registration does. An empty `log_dirs` and a `None`
297/// `cordoned_log_dirs` mean no change.
298#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
299pub struct BrokerRegistrationChangeRecord {
300    pub node_id: NodeId,
301    pub broker_epoch: i64,
302    pub fenced: FencingChange,
303    pub in_controlled_shutdown: bool,
304    /// KIP-858: the broker's online log directories after the change.
305    pub log_dirs: Vec<Uuid>,
306    /// KIP-1066: the broker's cordoned log directories after the change.
307    pub cordoned_log_dirs: Option<Vec<Uuid>>,
308}
309
310impl BrokerRegistrationChangeRecord {
311    /// A change that updates nothing, for `node_id` at `broker_epoch`. Set the
312    /// fields to change on the result.
313    #[must_use]
314    pub fn no_change(node_id: NodeId, broker_epoch: i64) -> Self {
315        Self {
316            node_id,
317            broker_epoch,
318            fenced: FencingChange::None,
319            in_controlled_shutdown: false,
320            log_dirs: Vec::new(),
321            cordoned_log_dirs: None,
322        }
323    }
324
325    /// The registration after this change applies to `current`, as Kafka's
326    /// `BrokerRegistration.cloneWith` builds it. The caller checks the epoch.
327    #[must_use]
328    pub fn applied_to(&self, current: &BrokerRegistrationRecord) -> BrokerRegistrationRecord {
329        BrokerRegistrationRecord {
330            fenced: self.fenced.apply(current.fenced),
331            in_controlled_shutdown: current.in_controlled_shutdown || self.in_controlled_shutdown,
332            log_dirs: if self.log_dirs.is_empty() {
333                current.log_dirs.clone()
334            } else {
335                self.log_dirs.clone()
336            },
337            cordoned_log_dirs: self
338                .cordoned_log_dirs
339                .clone()
340                .or_else(|| current.cordoned_log_dirs.clone()),
341            ..current.clone()
342        }
343    }
344}
345
346/// KIP-1312 `UnregisterController` (`api_key` 94), Kafka trunk's
347/// `UnregisterControllerRecord` (metadata apiKey 29).
348///
349/// Removes the registration of one controller. An apply against an unknown
350/// `node_id` does nothing.
351#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
352pub struct UnregisterControllerRecord {
353    pub node_id: NodeId,
354}
355
356/// Mutable topic configuration overrides.
357///
358/// Authoritative target state: each `V1TopicConfig` record fully replaces the
359/// previous override map for `topic`. An empty map clears all overrides. The
360/// `AlterConfigs` handler merges before it submits the record.
361#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
362pub struct TopicConfigRecord {
363    pub topic: String,
364    pub overrides: BTreeMap<String, String>,
365}
366
367/// Per-broker configuration key/value pair. `Some(value)` = set; `None` = delete.
368///
369/// Kafka names the cluster-wide broker-default resource with an empty string.
370/// [`DEFAULT_BROKER_CONFIG_NODE_ID`] represents that resource internally; the
371/// `KRaft` translator maps it back to the empty resource name.
372#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
373pub struct BrokerConfigRecord {
374    pub node_id: NodeId,
375    pub config_name: String,
376    /// `Some(value)` = set; `None` = delete.
377    pub config_value: Option<String>,
378}
379
380/// Internal identity for Kafka's cluster-wide default broker-config resource.
381/// Broker IDs are non-negative signed 32-bit integers on the Kafka wire, so
382/// this value cannot collide with a real broker.
383pub const DEFAULT_BROKER_CONFIG_NODE_ID: NodeId = NodeId(u64::MAX);
384
385/// KIP-714 client-metrics subscription config.
386///
387/// Authoritative target state: each `V1ClientMetricsConfig` fully replaces the
388/// previous override map for `name`, the subscription name. An empty map
389/// deletes the subscription. The `IncrementalAlterConfigs` handler merges
390/// before it submits the record, in the same pattern as [`TopicConfigRecord`].
391#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
392pub struct ClientMetricsConfigRecord {
393    pub name: String,
394    pub configs: BTreeMap<String, String>,
395}
396
397/// KIP-1071 dynamic configuration for one group resource. Each record is the
398/// authoritative override map for `group_id`; an empty map clears the resource.
399#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
400pub struct GroupConfigRecord {
401    pub group_id: String,
402    pub configs: BTreeMap<String, String>,
403}
404
405#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
406pub struct QuotaEntity {
407    pub entity_type: String,
408    pub entity_name: Option<String>,
409}
410
411#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
412pub struct ClientQuotaRecord {
413    /// Canonicalized entity tuple, sorted alphabetically by `entity_type`.
414    pub entity: Vec<QuotaEntity>,
415    pub config_key: String,
416    pub config_value: Option<f64>,
417}
418
419/// Durable controller state for cluster-wide producer-ID block allocation.
420/// `next_producer_id` is the first ID not covered by any committed block.
421#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
422pub struct ProducerIdsRecord {
423    pub broker_id: NodeId,
424    pub broker_epoch: i64,
425    pub next_producer_id: i64,
426}
427
428#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
429pub struct ScramCredentialRecord {
430    pub user: String,
431    pub mechanism: krabka_security::SaslMechanism,
432    pub salt: Vec<u8>,
433    pub stored_key: Vec<u8>,
434    pub server_key: Vec<u8>,
435    pub iterations: u32,
436}
437
438#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
439pub struct DeleteScramCredentialRecord {
440    pub user: String,
441    pub mechanism: krabka_security::SaslMechanism,
442}
443
444/// A single delegation token's authoritative state (KIP-48).
445///
446/// The record has replacement semantics: a new record with the same
447/// `token_id` overwrites the prior one in the image. Both Create and Renew
448/// use that. Removal goes through
449/// [`DeleteDelegationTokenRecord`].
450///
451/// The fields are those of Kafka's `DelegationTokenRecord`. Like Kafka's, the
452/// record carries no HMAC: the token's password is
453/// [`krabka_security::compute_token_hmac`] of `token_id` under the cluster's
454/// secret key, recomputed wherever it is needed.
455#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
456pub struct DelegationTokenRecord {
457    pub token_id: String,
458    pub owner: krabka_security::KafkaPrincipal,
459    /// The principal that created the token, which is the owner unless an
460    /// administrator created it on the owner's behalf (KIP-373).
461    pub requester: krabka_security::KafkaPrincipal,
462    pub issue_timestamp_ms: i64,
463    pub expiry_timestamp_ms: i64,
464    /// Issue plus max-lifetime. A renewal cannot push `expiry_timestamp_ms`
465    /// past this ceiling.
466    pub max_timestamp_ms: i64,
467    pub renewers: Vec<krabka_security::KafkaPrincipal>,
468}
469
470/// Tombstone record that removes a delegation token (KIP-48)
471/// from the image. The `ExpireDelegationToken` handlers and the
472/// background expiry sweep emit it.
473#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
474pub struct DeleteDelegationTokenRecord {
475    pub token_id: String,
476}
477
478/// KIP-853: finalizes the cluster-wide kraft.version feature level.
479#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
480pub struct KRaftVersionRecord {
481    pub kraft_version: u16,
482}
483
484/// KIP-853: full snapshot of the controller voter set.
485#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
486pub struct VotersRecord {
487    pub voters: VoterSet,
488}
489
490/// KIP-584 finalized feature level.
491///
492/// `level` is the finalized `max_version_level` for `name`. `level == 0` is the
493/// KIP-584 sentinel for "delete this finalized feature": `MetadataImage::apply`
494/// removes the entry and does not store a zero. Replacement semantics: a later
495/// record with the same `name` overwrites the previous level.
496#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
497pub struct FeatureLevelRecord {
498    pub name: String,
499    pub level: i16,
500}
501
502/// Snapshot-only carrier for the KIP-584 finalized-features epoch.
503///
504/// Apply normally derives the epoch, with one bump per applied
505/// `V1FeatureLevel`, so it tracks the history of `UpdateFeatures` calls and
506/// not the live feature count. That derivation cannot survive a snapshot. A
507/// snapshot stores resulting *state*, so it emits at most one
508/// `V1FeatureLevel` per live feature, which is fewer records than the
509/// original apply history. A replay of those records alone reconstructs a
510/// smaller epoch and diverges from a replica that replayed the full log.
511///
512/// [`MetadataImage::to_records`](crate::MetadataImage::to_records) therefore
513/// emits this record last, and
514/// [`MetadataImage::apply`](crate::MetadataImage::apply) SETS the epoch from
515/// it verbatim and does not bump it. That pins the reconstructed epoch to the
516/// original. Only `to_records` produces this record, and only snapshot replay
517/// consumes it. Nothing submits it as a controller change, so it never appears
518/// in the live Raft log.
519#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
520pub struct FeaturesEpochRecord {
521    pub epoch: i64,
522}
523
524#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
525#[non_exhaustive]
526pub enum MetadataRecord {
527    V1Topic(TopicRecord),
528    V1Partition(PartitionRecord),
529    V1BrokerRegistration(BrokerRegistrationRecord),
530    V1DeleteTopic(DeleteTopicRecord),
531    V1TopicConfig(TopicConfigRecord),
532    V1ScramCredential(ScramCredentialRecord),
533    V1DeleteScramCredential(DeleteScramCredentialRecord),
534    V1AccessControlEntry(crate::AclEntry),
535    V1DeleteAccessControlEntry(crate::AclEntryFilter),
536    V1BrokerConfig(BrokerConfigRecord),
537    V1ClientQuota(ClientQuotaRecord),
538    V1ProducerIds(ProducerIdsRecord),
539    V1DelegationToken(DelegationTokenRecord),
540    V1DeleteDelegationToken(DeleteDelegationTokenRecord),
541    V1UnregisterBroker(UnregisterBrokerRecord),
542    V1KRaftVersion(KRaftVersionRecord),
543    V1Voters(VotersRecord),
544    V1FeatureLevel(FeatureLevelRecord),
545    V1ClientMetricsConfig(ClientMetricsConfigRecord),
546    /// Snapshot-only: pins the finalized-features epoch on reconstruction.
547    /// Nothing submits it through the controller. See [`FeaturesEpochRecord`].
548    V1FeaturesEpoch(FeaturesEpochRecord),
549    /// KIP-858 directory-assignment delta (see [`PartitionDirAssignmentRecord`]).
550    /// Applied as a merge into one replica's `directories` slot; on the `KRaft`
551    /// log it rides a Krabka-private carrier so it stays a delta end-to-end.
552    V1PartitionDirAssignment(PartitionDirAssignmentRecord),
553    /// Diskless offset-sequencer delta (see [`PartitionOffsetAdvanceRecord`]).
554    /// Applied as an increment to the partition's committed next-offset.
555    V1PartitionOffsetAdvance(PartitionOffsetAdvanceRecord),
556    /// KIP-1071 dynamic GROUP resource configuration.
557    V1GroupConfig(GroupConfigRecord),
558    /// KIP-919 controller registration.
559    V1ControllerRegistration(ControllerRegistrationRecord),
560    /// One write-freeze registry entry (see
561    /// [`TopicFreezeRecord`](crate::write_freeze::TopicFreezeRecord)).
562    /// `frozen: false` removes the entry and is the thaw, so the registry
563    /// needs no separate tombstone record.
564    V1TopicFreeze(TopicFreezeRecord),
565    /// One break-glass proposal (see
566    /// [`BreakGlassProposalRecord`](crate::break_glass::BreakGlassProposalRecord)).
567    /// Replacement semantics on `proposal_id`: an approval and a consumption
568    /// each write the whole record back.
569    V1BreakGlassProposal(BreakGlassProposalRecord),
570    /// Tombstone that removes a break-glass proposal by id. The expiry sweep
571    /// emits it, in the same shape as `V1DeleteDelegationToken`.
572    V1DeleteBreakGlassProposal(Uuid),
573    /// KIP-966 per-partition eligible-leader state.
574    V1PartitionElr(PartitionElrRecord),
575    /// KIP-704 per-partition leader recovery state.
576    V1PartitionRecovery(PartitionRecoveryRecord),
577    /// Atomic standard-KRaft partition update decoded from a combined delta.
578    V1PartitionUpdate(PartitionUpdateRecord),
579    /// Kafka's `BrokerRegistrationChangeRecord`: a fencing, controlled-shutdown
580    /// or log-dir delta on one broker registration, applied only at the
581    /// registration's epoch.
582    V1BrokerRegistrationChange(BrokerRegistrationChangeRecord),
583    /// KIP-1312 controller unregistration.
584    V1UnregisterController(UnregisterControllerRecord),
585}
586
587#[cfg(test)]
588mod tests {
589
590    use serde_wincode::SerdeCompat;
591    use wincode::{Deserialize as _, Serialize as _};
592
593    use super::*;
594    use crate::{
595        break_glass::{BreakGlassAction, BreakGlassApproval},
596        voters::{KRaftVersionRange, Voter, VoterEndpoint},
597    };
598
599    fn round_trip(r: &MetadataRecord) -> MetadataRecord {
600        let bytes = <SerdeCompat<MetadataRecord>>::serialize(r).unwrap();
601        <SerdeCompat<MetadataRecord>>::deserialize(&bytes).unwrap()
602    }
603
604    #[test]
605    fn feature_level_round_trip() {
606        let r = MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
607            name: "metadata.version".into(),
608            level: 1,
609        });
610        assert2::assert!(round_trip(&r) == r);
611    }
612
613    #[test]
614    fn group_config_round_trip() {
615        let r = MetadataRecord::V1GroupConfig(GroupConfigRecord {
616            group_id: "streams-app".into(),
617            configs: BTreeMap::from([("streams.num.standby.replicas".into(), "1".into())]),
618        });
619        assert2::assert!(round_trip(&r) == r);
620    }
621
622    #[test]
623    fn controller_registration_round_trip() {
624        let r = MetadataRecord::V1ControllerRegistration(ControllerRegistrationRecord {
625            node_id: NodeId(3),
626            incarnation_id: Uuid::from_u128(7),
627            zk_migration_ready: false,
628            endpoints: vec![BrokerEndpoint {
629                name: "CONTROLLER".into(),
630                host: "controller-3".into(),
631                port: 9093,
632                protocol: krabka_security::ListenerProtocol::Plaintext,
633            }],
634            features: BTreeMap::from([("metadata.version".into(), (7, 25))]),
635        });
636        assert2::assert!(round_trip(&r) == r);
637    }
638
639    #[test]
640    fn producer_ids_round_trip() {
641        let r = MetadataRecord::V1ProducerIds(ProducerIdsRecord {
642            broker_id: NodeId(3),
643            broker_epoch: 9,
644            next_producer_id: 2_000,
645        });
646        assert2::assert!(round_trip(&r) == r);
647    }
648
649    #[test]
650    fn features_epoch_round_trip() {
651        let r = MetadataRecord::V1FeaturesEpoch(FeaturesEpochRecord { epoch: 7 });
652        assert2::assert!(round_trip(&r) == r);
653    }
654
655    #[test]
656    fn topic_record_round_trip() {
657        let r = MetadataRecord::V1Topic(TopicRecord {
658            name: "t".into(),
659            topic_id: Uuid::new_v4(),
660            partitions: 3,
661            replication_factor: 1,
662        });
663        assert2::assert!(round_trip(&r) == r);
664    }
665
666    #[test]
667    fn partition_record_round_trip() {
668        let r = MetadataRecord::V1Partition(PartitionRecord {
669            topic: "t".into(),
670            partition: 0,
671            leader: NodeId(1),
672            replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
673            isr: vec![NodeId(1), NodeId(2)],
674            leader_epoch: LeaderEpoch(0),
675            adding_replicas: vec![],
676            removing_replicas: vec![],
677            directories: vec![Uuid::from_u128(1), Uuid::from_u128(2), Uuid::nil()],
678            partition_epoch: 0,
679        });
680        assert2::assert!(round_trip(&r) == r);
681    }
682
683    #[test]
684    fn partition_dir_assignment_round_trip() {
685        let r = MetadataRecord::V1PartitionDirAssignment(PartitionDirAssignmentRecord {
686            topic: "t".into(),
687            partition: 2,
688            replica: NodeId(3),
689            directory: Uuid::from_u128(0xAB),
690        });
691        assert2::assert!(round_trip(&r) == r);
692    }
693
694    #[test]
695    fn broker_registration_round_trip() {
696        let r = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
697            node_id: NodeId(7),
698            broker_epoch: 0,
699            incarnation_id: Uuid::from_u128(0xdeadbeef_cafe_babe_0123_456789abcdef),
700            host: "192.168.1.10".into(),
701            port: 9092,
702            rack: Some("us-east-1a".into()),
703            log_dirs: vec![],
704            fenced: false,
705            in_controlled_shutdown: false,
706            cordoned_log_dirs: None,
707            endpoints: vec![],
708            features: BTreeMap::new(),
709        });
710        assert2::assert!(round_trip(&r) == r);
711    }
712
713    #[test]
714    fn broker_registration_with_endpoints_round_trip() {
715        let r = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
716            node_id: NodeId(1),
717            broker_epoch: 0,
718            incarnation_id: Uuid::from_u128(0xfeedface_0000_0000_0000_000000000001),
719            host: "h".into(),
720            port: 9092,
721            rack: None,
722            log_dirs: vec![],
723            fenced: false,
724            in_controlled_shutdown: false,
725            cordoned_log_dirs: None,
726            endpoints: vec![BrokerEndpoint {
727                name: "EXTERNAL".into(),
728                host: "ext.example.com".into(),
729                port: 9092,
730                protocol: krabka_security::ListenerProtocol::SaslSsl,
731            }],
732            features: BTreeMap::new(),
733        });
734        assert2::assert!(round_trip(&r) == r);
735    }
736
737    #[test]
738    fn delete_topic_round_trip() {
739        let r = MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
740            name: "doomed".into(),
741        });
742        assert2::assert!(round_trip(&r) == r);
743    }
744
745    #[test]
746    fn unregister_broker_round_trip() {
747        let r = MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
748            node_id: NodeId(42),
749            broker_epoch: 17,
750        });
751        assert2::assert!(round_trip(&r) == r);
752    }
753
754    #[test]
755    fn broker_registration_change_and_unregister_controller_round_trip() {
756        for r in [
757            MetadataRecord::V1BrokerRegistrationChange(BrokerRegistrationChangeRecord {
758                node_id: NodeId(4),
759                broker_epoch: 12,
760                fenced: FencingChange::Unfence,
761                in_controlled_shutdown: true,
762                log_dirs: vec![Uuid::from_u128(0xD1)],
763                cordoned_log_dirs: Some(vec![]),
764            }),
765            MetadataRecord::V1UnregisterController(UnregisterControllerRecord {
766                node_id: NodeId(3000),
767            }),
768        ] {
769            assert2::assert!(round_trip(&r) == r);
770        }
771    }
772
773    /// `FencingChange` is Kafka's `BrokerRegistrationFencingChange`: -1
774    /// unfences, 0 leaves the state alone and 1 fences.
775    #[test]
776    fn fencing_change_matches_kafka() {
777        for (change, wire, from_false, from_true) in [
778            (FencingChange::Unfence, -1, false, false),
779            (FencingChange::None, 0, false, true),
780            (FencingChange::Fence, 1, true, true),
781        ] {
782            assert2::check!(
783                (
784                    change.wire_value(),
785                    FencingChange::from_wire(wire),
786                    change.apply(false),
787                    change.apply(true),
788                ) == (wire, Some(change), from_false, from_true)
789            );
790        }
791        for undefined in [-2, 2, i8::MIN, i8::MAX] {
792            assert2::check!(FencingChange::from_wire(undefined) == None);
793        }
794    }
795
796    /// `applied_to` is Kafka's `BrokerRegistration.cloneWith`: each field of
797    /// the change either leaves the registration's value or replaces it, and
798    /// controlled shutdown can only turn on.
799    #[test]
800    fn registration_change_applies_like_clone_with() {
801        let (d1, d2) = (Uuid::from_u128(0xD1), Uuid::from_u128(0xD2));
802        let current = BrokerRegistrationRecord {
803            node_id: NodeId(4),
804            broker_epoch: 12,
805            incarnation_id: Uuid::from_u128(4),
806            host: "broker-4".into(),
807            port: 9092,
808            rack: None,
809            endpoints: vec![],
810            log_dirs: vec![d1],
811            fenced: true,
812            in_controlled_shutdown: false,
813            cordoned_log_dirs: None,
814            features: BTreeMap::new(),
815        };
816        let base = BrokerRegistrationChangeRecord::no_change(NodeId(4), 12);
817        let with = |edit: fn(&mut BrokerRegistrationRecord)| {
818            let mut want = current.clone();
819            edit(&mut want);
820            want
821        };
822        for (case, change, want) in [
823            ("no change", base.clone(), current.clone()),
824            (
825                "unfence",
826                BrokerRegistrationChangeRecord {
827                    fenced: FencingChange::Unfence,
828                    ..base.clone()
829                },
830                with(|b| b.fenced = false),
831            ),
832            (
833                "controlled shutdown",
834                BrokerRegistrationChangeRecord {
835                    in_controlled_shutdown: true,
836                    ..base.clone()
837                },
838                with(|b| b.in_controlled_shutdown = true),
839            ),
840            (
841                "new log dirs",
842                BrokerRegistrationChangeRecord {
843                    log_dirs: vec![d2],
844                    ..base.clone()
845                },
846                with(|b| b.log_dirs = vec![Uuid::from_u128(0xD2)]),
847            ),
848            (
849                "uncordon every dir",
850                BrokerRegistrationChangeRecord {
851                    cordoned_log_dirs: Some(vec![]),
852                    ..base.clone()
853                },
854                with(|b| b.cordoned_log_dirs = Some(vec![])),
855            ),
856        ] {
857            assert2::check!(change.applied_to(&current) == want, "{case}");
858        }
859
860        let shutting_down = BrokerRegistrationRecord {
861            in_controlled_shutdown: true,
862            ..current.clone()
863        };
864        assert2::check!(base.applied_to(&shutting_down) == shutting_down);
865    }
866
867    #[test]
868    fn topic_config_record_round_trip() {
869        let mut overrides = BTreeMap::new();
870        overrides.insert("retention.ms".to_owned(), "60000".to_owned());
871        overrides.insert("segment.bytes".to_owned(), "1048576".to_owned());
872        let r = MetadataRecord::V1TopicConfig(TopicConfigRecord {
873            topic: "t".into(),
874            overrides,
875        });
876        assert2::assert!(round_trip(&r) == r);
877    }
878
879    #[test]
880    fn scram_credential_round_trip() {
881        let r = MetadataRecord::V1ScramCredential(ScramCredentialRecord {
882            user: "alice".into(),
883            mechanism: krabka_security::SaslMechanism::ScramSha512,
884            salt: vec![1u8; 16],
885            stored_key: vec![2u8; 64],
886            server_key: vec![3u8; 64],
887            iterations: 4096,
888        });
889        assert2::assert!(round_trip(&r) == r);
890    }
891
892    #[test]
893    fn delete_scram_credential_round_trip() {
894        let r = MetadataRecord::V1DeleteScramCredential(DeleteScramCredentialRecord {
895            user: "alice".into(),
896            mechanism: krabka_security::SaslMechanism::ScramSha512,
897        });
898        assert2::assert!(round_trip(&r) == r);
899    }
900
901    #[test]
902    fn v1_access_control_entry_round_trip() {
903        let entry = crate::AclEntry {
904            resource_type: crate::ResourceType::Topic,
905            resource_name: "foo".into(),
906            pattern_type: crate::PatternType::Literal,
907            principal: "User:alice".into(),
908            host: "*".into(),
909            operation: crate::AclOperation::Read,
910            permission_type: crate::PermissionType::Allow,
911        };
912        let r = MetadataRecord::V1AccessControlEntry(entry);
913        assert2::assert!(round_trip(&r) == r);
914    }
915
916    #[test]
917    fn v1_delete_access_control_entry_round_trip() {
918        let filter = crate::AclEntryFilter {
919            resource_type: Some(crate::ResourceType::Group),
920            resource_name: Some("cg-foo".into()),
921            pattern_type: Some(crate::PatternType::Literal),
922            principal: None,
923            host: None,
924            operation: None,
925            permission_type: None,
926        };
927        let r = MetadataRecord::V1DeleteAccessControlEntry(filter);
928        assert2::assert!(round_trip(&r) == r);
929    }
930
931    #[test]
932    fn broker_config_record_round_trip() {
933        let r = MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
934            node_id: NodeId(7),
935            config_name: "leader.replication.throttled.rate".into(),
936            config_value: Some("2048".into()),
937        });
938        assert2::assert!(round_trip(&r) == r);
939    }
940
941    #[test]
942    fn client_quota_record_round_trip() {
943        let r = MetadataRecord::V1ClientQuota(ClientQuotaRecord {
944            entity: vec![
945                QuotaEntity {
946                    entity_type: "client-id".into(),
947                    entity_name: Some("app1".into()),
948                },
949                QuotaEntity {
950                    entity_type: "user".into(),
951                    entity_name: Some("alice".into()),
952                },
953            ],
954            config_key: "producer_byte_rate".into(),
955            config_value: Some(1024.0),
956        });
957        assert2::assert!(round_trip(&r) == r);
958    }
959
960    #[test]
961    fn delegation_token_record_round_trip() {
962        let r = MetadataRecord::V1DelegationToken(DelegationTokenRecord {
963            token_id: "tok-abc".into(),
964            owner: krabka_security::KafkaPrincipal {
965                principal_type: "User".into(),
966                name: "alice".into(),
967            },
968            requester: krabka_security::KafkaPrincipal {
969                principal_type: "User".into(),
970                name: "admin".into(),
971            },
972            issue_timestamp_ms: 1_700_000_000_000,
973            expiry_timestamp_ms: 1_700_000_600_000,
974            max_timestamp_ms: 1_700_604_800_000,
975            renewers: vec![krabka_security::KafkaPrincipal {
976                principal_type: "User".into(),
977                name: "bob".into(),
978            }],
979        });
980        assert2::assert!(round_trip(&r) == r);
981    }
982
983    #[test]
984    fn delete_delegation_token_record_round_trip() {
985        let r = MetadataRecord::V1DeleteDelegationToken(DeleteDelegationTokenRecord {
986            token_id: "tok-abc".into(),
987        });
988        assert2::assert!(round_trip(&r) == r);
989    }
990
991    #[test]
992    fn voters_record_round_trips() {
993        let rec = MetadataRecord::V1Voters(VotersRecord {
994            voters: VoterSet::from_voters([Voter {
995                id: NodeId(7),
996                directory_id: uuid::Uuid::from_u128(7),
997                endpoints: vec![VoterEndpoint {
998                    name: "CONTROLLER".into(),
999                    host: "h".into(),
1000                    port: 1,
1001                }],
1002                kraft_version: KRaftVersionRange::default(),
1003            }]),
1004        });
1005        assert2::assert!(round_trip(&rec) == rec);
1006    }
1007
1008    #[test]
1009    fn kraft_version_record_round_trips() {
1010        let rec = MetadataRecord::V1KRaftVersion(KRaftVersionRecord { kraft_version: 1 });
1011        assert2::assert!(round_trip(&rec) == rec);
1012    }
1013
1014    #[test]
1015    fn client_metrics_config_round_trip() {
1016        let mut overrides = BTreeMap::new();
1017        overrides.insert("interval.ms".to_owned(), "60000".to_owned());
1018        overrides.insert(
1019            "metrics".to_owned(),
1020            "org.apache.kafka.consumer.".to_owned(),
1021        );
1022        let r = MetadataRecord::V1ClientMetricsConfig(ClientMetricsConfigRecord {
1023            name: "sub-a".into(),
1024            configs: overrides,
1025        });
1026        assert2::assert!(round_trip(&r) == r);
1027    }
1028
1029    #[test]
1030    fn topic_freeze_round_trip() {
1031        let r = MetadataRecord::V1TopicFreeze(TopicFreezeRecord {
1032            scope: "tenant-a.".into(),
1033            pattern_type: crate::PatternType::Prefixed,
1034            frozen: true,
1035            reason: "DR cutover".into(),
1036            set_by: "User:alice".into(),
1037            set_at_ms: 1_700_000_000_000,
1038            proposal_id: Uuid::from_u128(0xB1),
1039            key_id: "alice-yubi".into(),
1040            signature: vec![0xAB; 64],
1041        });
1042        assert2::assert!(round_trip(&r) == r);
1043    }
1044
1045    #[test]
1046    fn break_glass_proposal_round_trip() {
1047        let r = MetadataRecord::V1BreakGlassProposal(BreakGlassProposalRecord {
1048            proposal_id: Uuid::from_u128(0xB1),
1049            action: BreakGlassAction::ThawTopicFreeze,
1050            target: "literal:orders".into(),
1051            proposer: "User:alice".into(),
1052            reason: "incident 4711 closed".into(),
1053            created_at_ms: 1_700_000_000_000,
1054            expires_at_ms: 1_700_000_600_000,
1055            approvals: vec![BreakGlassApproval {
1056                principal: "User:bob".into(),
1057                approved_at_ms: 1_700_000_060_000,
1058                key_id: "bob-yubi".into(),
1059                signature: vec![0xCD; 64],
1060            }],
1061            consumed_at_ms: 0,
1062            withdrawn: false,
1063        });
1064        assert2::assert!(round_trip(&r) == r);
1065    }
1066
1067    #[test]
1068    fn delete_break_glass_proposal_round_trip() {
1069        let r = MetadataRecord::V1DeleteBreakGlassProposal(Uuid::from_u128(0xB1));
1070        assert2::assert!(round_trip(&r) == r);
1071    }
1072
1073    #[test]
1074    fn partition_epoch_serde_default_is_minus_one() {
1075        // -1 is Kafka's "unknown epoch" sentinel; pin it (mutants flip it to
1076        // 0/1). This is the `#[serde(default)]` fallback for partition_epoch.
1077        assert2::assert!(default_partition_epoch() == -1);
1078    }
1079}
krabka_metadata/records.rs

krabka_metadata/
records.rs
