krabka_raft/
voter_requests.rs

1//! The KIP-853 request checks that `AddRaftVoter`, `RemoveRaftVoter` and
2//! `UpdateRaftVoter` share on the controller listener and on the broker
3//! listener, with Kafka's codes, order and messages.
4//!
5//! The order is that of `KafkaRaftClient.handleAddVoterRequest`,
6//! `handleRemoveVoterRequest` and `handleUpdateVoterRequest`: the cluster id,
7//! then the leader, then the voter key, then the endpoints. Each function
8//! answers `None` when the request passes, so the caller goes on to the
9//! reconfiguration itself.
10
11use {
12    crate::{RaftError, kraft::transport::QuorumStateSnapshot, reconfig::ReconfigOutcome},
13    base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD},
14    krabka_protocol::{
15        owned::{
16            add_raft_voter_request::AddRaftVoterRequest,
17            remove_raft_voter_request::RemoveRaftVoterRequest,
18            update_raft_voter_request::UpdateRaftVoterRequest,
19            update_raft_voter_response::CurrentLeader,
20        },
21        primitives::uuid::Uuid as WireUuid,
22    },
23    std::collections::BTreeSet,
24};
25
26/// Kafka's `NOT_LEADER_OR_FOLLOWER`.
27pub const NOT_LEADER_OR_FOLLOWER: i16 = 6;
28/// Kafka's `REQUEST_TIMED_OUT`.
29pub const REQUEST_TIMED_OUT: i16 = 7;
30/// Kafka's `UNSUPPORTED_VERSION`.
31pub const UNSUPPORTED_VERSION: i16 = 35;
32/// Kafka's `INVALID_REQUEST`.
33pub const INVALID_REQUEST: i16 = 42;
34/// Kafka's `FENCED_LEADER_EPOCH`.
35pub const FENCED_LEADER_EPOCH: i16 = 74;
36/// Kafka's `UNKNOWN_LEADER_EPOCH`.
37pub const UNKNOWN_LEADER_EPOCH: i16 = 75;
38/// Kafka's `INCONSISTENT_CLUSTER_ID`.
39pub const INCONSISTENT_CLUSTER_ID: i16 = 104;
40/// Kafka's `DUPLICATE_VOTER`.
41pub const DUPLICATE_VOTER: i16 = 126;
42/// Kafka's `VOTER_NOT_FOUND`.
43pub const VOTER_NOT_FOUND: i16 = 127;
44/// Kafka's `UNKNOWN_SERVER_ERROR`.
45pub const UNKNOWN_SERVER_ERROR: i16 = -1;
46
47/// The listener name a controller advertises its peer RPCs on, by the
48/// convention of `controller_endpoint_addr`.
49const CONTROLLER_LISTENER_NAME: &str = "CONTROLLER";
50
51/// An error code and the message beside it.
52pub type Refusal = (i16, Option<String>);
53
54/// Kafka's `ReplicaKey.toString`, with the directory id in Kafka's base64
55/// `Uuid` form.
56#[must_use]
57pub fn replica_key(voter_id: i32, directory_id: WireUuid) -> String {
58    let directory = if directory_id == WireUuid::ZERO {
59        "<undefined>".to_owned()
60    } else {
61        URL_SAFE_NO_PAD.encode(directory_id.0)
62    };
63    format!("ReplicaKey(id={voter_id}, directoryId={directory})")
64}
65
66/// Whether a request's cluster id names this cluster. `cluster_id` is
67/// `java.util.UUID`'s hyphenated form, while krabka now reports its cluster
68/// id in Kafka's base64 `Uuid` form (`Metadata`, `DescribeCluster`; see
69/// #1042), so a caller may send either form of the same id, and both match.
70fn names_this_cluster(request_cluster_id: &str, cluster_id: &str) -> bool {
71    request_cluster_id == cluster_id
72        || uuid::Uuid::parse_str(cluster_id)
73            .is_ok_and(|id| URL_SAFE_NO_PAD.encode(id.as_bytes()) == request_cluster_id)
74}
75
76/// Kafka's `hasValidClusterId` refusal, with its message. An absent id is
77/// valid.
78fn cluster_id_refusal(request_cluster_id: Option<&str>, cluster_id: &str) -> Option<Refusal> {
79    let request_cluster_id = request_cluster_id.filter(|id| !names_this_cluster(id, cluster_id))?;
80    Some((
81        INCONSISTENT_CLUSTER_ID,
82        Some(format!(
83            "The given id \"{request_cluster_id}\" doesn't match the cluster id \"{cluster_id}\""
84        )),
85    ))
86}
87
88/// The name of the leader's controller listener: Kafka's
89/// `channel.listenerName()` on the leader, which a new or updated voter must
90/// advertise. It is `None` when this node does not know its own endpoints.
91#[must_use]
92pub fn leader_listener_name(quorum: &QuorumStateSnapshot) -> Option<String> {
93    let endpoints = &quorum.voters.get(quorum.leader_id?)?.endpoints;
94    endpoints
95        .iter()
96        .find(|endpoint| endpoint.name.eq_ignore_ascii_case(CONTROLLER_LISTENER_NAME))
97        .or_else(|| endpoints.first())
98        .map(|endpoint| endpoint.name.clone())
99}
100
101/// Whether `listeners` carry the leader's controller listener name. Kafka's
102/// `ListenerName.normalised` makes the match case-insensitive.
103fn carries_leader_listener<'a>(
104    quorum: &QuorumStateSnapshot,
105    mut names: impl Iterator<Item = &'a str>,
106) -> bool {
107    leader_listener_name(quorum)
108        .is_none_or(|leader| names.any(|name| name.eq_ignore_ascii_case(&leader)))
109}
110
111/// A listener as a request names it on the wire, borrowed from the request.
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub struct ListenerAddress<'a> {
114    /// The listener name.
115    pub name: &'a str,
116    /// The host the listener is reached at.
117    pub host: &'a str,
118    /// The port the listener is reached at.
119    pub port: u16,
120}
121
122/// A listener set is usable when every entry is named, hosted and on a real
123/// port, no name repeats, and there is at least one.
124#[must_use]
125pub fn valid_wire_listeners<'a>(listeners: impl IntoIterator<Item = ListenerAddress<'a>>) -> bool {
126    let mut names = BTreeSet::new();
127    let mut count = 0usize;
128    for ListenerAddress { name, host, port } in listeners {
129        count += 1;
130        if name.is_empty() || host.is_empty() || port == 0 || !names.insert(name) {
131            return false;
132        }
133    }
134    count != 0
135}
136
137fn valid_voter_identity<'a>(
138    voter_id: i32,
139    directory_id: WireUuid,
140    listeners: impl IntoIterator<Item = ListenerAddress<'a>>,
141) -> bool {
142    voter_id >= 0 && directory_id != WireUuid::ZERO && valid_wire_listeners(listeners)
143}
144
145macro_rules! request_identity_is_valid {
146    ($request:expr) => {
147        valid_voter_identity(
148            $request.voter_id,
149            $request.voter_directory_id,
150            $request.listeners.iter().map(|listener| ListenerAddress {
151                name: listener.name.as_str(),
152                host: listener.host.as_str(),
153                port: listener.port,
154            }),
155        )
156    };
157}
158
159fn leader_request_refusal(
160    request_cluster: Option<&str>,
161    cluster_id: &str,
162    quorum: &QuorumStateSnapshot,
163) -> Option<Refusal> {
164    cluster_id_refusal(request_cluster, cluster_id)
165        .or_else(|| (!quorum.is_leader).then_some((NOT_LEADER_OR_FOLLOWER, None)))
166}
167
168fn voter_identity_refusal(
169    request_cluster: Option<&str>,
170    cluster_id: &str,
171    quorum: &QuorumStateSnapshot,
172    valid: impl FnOnce() -> bool,
173    invalid_message: &str,
174) -> Option<Refusal> {
175    leader_request_refusal(request_cluster, cluster_id, quorum)
176        .or_else(|| (!valid()).then(|| (INVALID_REQUEST, Some(invalid_message.into()))))
177}
178
179/// The refusal of an `AddRaftVoter` request before the candidate probe, as
180/// `KafkaRaftClient.handleAddVoterRequest` orders it.
181#[must_use]
182pub fn add_voter_refusal(
183    request: &AddRaftVoterRequest,
184    cluster_id: &str,
185    quorum: &QuorumStateSnapshot,
186) -> Option<Refusal> {
187    if let Some(refusal) = voter_identity_refusal(
188        request.cluster_id.as_deref(),
189        cluster_id,
190        quorum,
191        || request_identity_is_valid!(request),
192        "Add voter request didn't include a valid voter",
193    ) {
194        return Some(refusal);
195    }
196    if !carries_leader_listener(
197        quorum,
198        request
199            .listeners
200            .iter()
201            .map(|listener| listener.name.as_str()),
202    ) {
203        return Some((
204            INVALID_REQUEST,
205            Some(format!(
206                "Add voter request didn't include the endpoint for the default listener {}",
207                leader_listener_name(quorum).unwrap_or_default()
208            )),
209        ));
210    }
211    None
212}
213
214/// The refusal of a candidate whose `kraft.version` range does not cover the
215/// finalized version: `INVALID_REQUEST`, as `AddVoterHandler` answers it.
216#[must_use]
217pub fn candidate_kraft_version_refusal(
218    voter_id: i32,
219    directory_id: WireUuid,
220    finalized_version: u16,
221) -> Refusal {
222    (
223        INVALID_REQUEST,
224        Some(format!(
225            "Aborted add voter operation for {} since the kraft.version range doesn't support \
226             the finalized version {finalized_version}",
227            replica_key(voter_id, directory_id)
228        )),
229    )
230}
231
232/// The refusal of a candidate that could not be asked for its
233/// `ApiVersions`: `REQUEST_TIMED_OUT`, as `AddVoterHandler` answers it.
234#[must_use]
235pub fn candidate_unavailable_refusal(
236    voter_id: i32,
237    directory_id: WireUuid,
238    error: &str,
239) -> Refusal {
240    (
241        REQUEST_TIMED_OUT,
242        Some(format!(
243            "Aborted add voter operation for {} since API_VERSIONS returned an error {error}",
244            replica_key(voter_id, directory_id)
245        )),
246    )
247}
248
249/// The refusal of a `RemoveRaftVoter` request before the reconfiguration, as
250/// `KafkaRaftClient.handleRemoveVoterRequest` orders it.
251#[must_use]
252pub fn remove_voter_refusal(
253    request: &RemoveRaftVoterRequest,
254    cluster_id: &str,
255    quorum: &QuorumStateSnapshot,
256) -> Option<Refusal> {
257    voter_identity_refusal(
258        request.cluster_id.as_deref(),
259        cluster_id,
260        quorum,
261        || request.voter_id >= 0 && request.voter_directory_id != WireUuid::ZERO,
262        "Remove voter request didn't include a valid voter",
263    )
264}
265
266/// The refusal code of an `UpdateRaftVoter` request before the
267/// reconfiguration, as `KafkaRaftClient.handleUpdateVoterRequest` and
268/// `UpdateVoterHandler` order it. The response carries no message.
269#[must_use]
270pub fn update_voter_refusal(
271    request: &UpdateRaftVoterRequest,
272    cluster_id: &str,
273    quorum: &QuorumStateSnapshot,
274) -> Option<i16> {
275    if cluster_id_refusal(request.cluster_id.as_deref(), cluster_id).is_some() {
276        return Some(INCONSISTENT_CLUSTER_ID);
277    }
278    let local_epoch = i64::from(quorum.leader_epoch);
279    let request_epoch = i64::from(request.current_leader_epoch);
280    if request_epoch < local_epoch {
281        return Some(FENCED_LEADER_EPOCH);
282    }
283    if request_epoch > local_epoch {
284        return Some(UNKNOWN_LEADER_EPOCH);
285    }
286    if !quorum.is_leader {
287        return Some(NOT_LEADER_OR_FOLLOWER);
288    }
289    let feature = &request.k_raft_version_feature;
290    if !request_identity_is_valid!(request)
291        || feature.min_supported_version < 0
292        || feature.max_supported_version < feature.min_supported_version
293        || !carries_leader_listener(
294            quorum,
295            request
296                .listeners
297                .iter()
298                .map(|listener| listener.name.as_str()),
299        )
300    {
301        return Some(INVALID_REQUEST);
302    }
303    None
304}
305
306/// `UpdateRaftVoterResponse.CurrentLeader`, as `RaftUtil.updateVoterResponse`
307/// fills it in every answer.
308///
309/// It holds the leader id or -1, the epoch, and the leader's
310/// host and port on the controller listener when this node knows them.
311#[must_use]
312pub fn update_voter_current_leader(quorum: &QuorumStateSnapshot) -> CurrentLeader {
313    let endpoint = quorum.leader_id.and_then(|leader| {
314        let name = leader_listener_name(quorum)?;
315        quorum
316            .voters
317            .get(leader)?
318            .endpoints
319            .iter()
320            .find(|endpoint| endpoint.name == name)
321            .cloned()
322    });
323    CurrentLeader {
324        leader_id: quorum
325            .leader_id
326            .map_or(-1, |leader| i32::try_from(leader.0).unwrap_or(-1)),
327        leader_epoch: i32::try_from(quorum.leader_epoch).unwrap_or(i32::MAX),
328        host: endpoint
329            .as_ref()
330            .map_or_default(|endpoint| endpoint.host.clone()),
331        port: endpoint.map_or(0, |endpoint| i32::from(endpoint.port)),
332        ..Default::default()
333    }
334}
335
336/// The voter operation a request asked for, which its refusal names as Kafka's
337/// `AddVoterHandler` and `RemoveVoterHandler` do.
338#[derive(Debug, Clone, Copy, PartialEq, Eq, krabka_macros::EnumStr)]
339#[enum_str(as_str = gerund)]
340pub enum VoterOperation {
341    #[enum_str(name = "adding")]
342    Add,
343    #[enum_str(name = "removing")]
344    Remove,
345    #[enum_str(name = "updating")]
346    Update,
347}
348
349/// The code and message of a reconfiguration's outcome, for the voter
350/// `(voter_id, directory_id)` the request named.
351#[must_use]
352pub fn reconfiguration_refusal(
353    result: Result<ReconfigOutcome, RaftError>,
354    operation: VoterOperation,
355    voter_id: i32,
356    directory_id: WireUuid,
357) -> Refusal {
358    match result {
359        Ok(ReconfigOutcome::Committed) => (0, None),
360        Ok(ReconfigOutcome::NotLeader { .. }) | Err(RaftError::NotLeader { .. }) => {
361            (NOT_LEADER_OR_FOLLOWER, None)
362        }
363        Err(RaftError::ReconfigInProgress) => (
364            REQUEST_TIMED_OUT,
365            Some(
366                "Request timed out waiting for leader to handle previous voter change request"
367                    .into(),
368            ),
369        ),
370        // `AddVoterHandler` aborts a candidate that is not caught up with
371        // `REQUEST_TIMED_OUT`, so the operator retries once it has fetched.
372        Err(RaftError::VoterNotCaughtUp { .. }) => (
373            REQUEST_TIMED_OUT,
374            Some(format!(
375                "Aborted add voter operation for {} since it is lagging behind",
376                replica_key(voter_id, directory_id)
377            )),
378        ),
379        Err(RaftError::DuplicateVoter(_)) => (
380            DUPLICATE_VOTER,
381            Some(format!(
382                "The voter id for {} is already part of the set of voters",
383                replica_key(voter_id, directory_id)
384            )),
385        ),
386        Err(RaftError::VoterNotFound(_)) => (
387            VOTER_NOT_FOUND,
388            Some(format!(
389                "Cannot remove voter {} from the set of voters",
390                replica_key(voter_id, directory_id)
391            )),
392        ),
393        Err(RaftError::UnsupportedKraftVersion(version)) => (
394            UNSUPPORTED_VERSION,
395            Some(format!(
396                "Cluster doesn't support {} voter because the kraft.version feature is {version}",
397                operation.gerund()
398            )),
399        ),
400        // Kafka has no "invalid voter update" code, so a rejected change and a
401        // malformed one land on the same `INVALID_REQUEST` that
402        // `UpdateVoterHandler` and `KafkaRaftClient` return.
403        Err(RaftError::InvalidVoterUpdate(message) | RaftError::ReconfigRejected(message)) => {
404            (INVALID_REQUEST, Some(message))
405        }
406        Err(error) => (UNKNOWN_SERVER_ERROR, Some(error.to_string())),
407    }
408}
409
410#[cfg(test)]
411mod tests {
412    use {
413        super::*,
414        crate::NodeId,
415        assert2::check,
416        krabka_protocol::owned::{
417            add_raft_voter_request::Listener as AddListener,
418            update_raft_voter_request::{KRaftVersionFeature, Listener as UpdateListener},
419        },
420        std::collections::BTreeMap,
421    };
422
423    const CLUSTER: &str = "cluster-a";
424    const DIRECTORY: WireUuid = WireUuid([9; 16]);
425    const TEST_VOTER_KEY: &str = "ReplicaKey(id=3, directoryId=CQkJCQkJCQkJCQkJCQkJCQ)";
426
427    /// Node 1 leading epoch 7 over voters 1 and 2, when `is_leader`, or node 2
428    /// following it.
429    fn quorum(is_leader: bool) -> QuorumStateSnapshot {
430        let voter = |id: u64, port: u16| krabka_metadata::Voter {
431            id: NodeId(id),
432            directory_id: uuid::Uuid::from_u128(u128::from(id)),
433            endpoints: vec![
434                krabka_metadata::VoterEndpoint {
435                    name: "PLAINTEXT".into(),
436                    host: "broker-host".into(),
437                    port: port + 1,
438                },
439                krabka_metadata::VoterEndpoint {
440                    name: "CONTROLLER".into(),
441                    host: format!("controller-{id}"),
442                    port,
443                },
444            ],
445            kraft_version: krabka_metadata::KRaftVersionRange::default(),
446        };
447        QuorumStateSnapshot {
448            leader_id: Some(NodeId(1)),
449            leader_epoch: 7,
450            high_watermark: 0,
451            last_applied_record_timestamp_ms: 0,
452            pending_record_processing_time_us_avg: 0,
453            pending_record_processing_time_us_max: 0,
454            quorum_high_watermark: 0,
455            log_end_offset: 0,
456            log_start_offset: 0,
457            voters: krabka_metadata::VoterSet::from_voters([voter(1, 9093), voter(2, 9094)]),
458            voted_directory_id: None,
459            observers: Vec::new(),
460            per_replica_fetch_offset: BTreeMap::new(),
461            per_replica_last_fetch_ms: BTreeMap::new(),
462            per_replica_last_caught_up_ms: BTreeMap::new(),
463            is_leader,
464            current_state: if is_leader { "leader" } else { "follower" },
465        }
466    }
467
468    fn foreign_cluster_refusal() -> Refusal {
469        (
470            INCONSISTENT_CLUSTER_ID,
471            Some("The given id \"cluster-b\" doesn't match the cluster id \"cluster-a\"".into()),
472        )
473    }
474
475    fn add_voter_request() -> AddRaftVoterRequest {
476        AddRaftVoterRequest {
477            cluster_id: None,
478            voter_id: 3,
479            voter_directory_id: DIRECTORY,
480            listeners: vec![AddListener {
481                name: "controller".into(),
482                host: "host-3".into(),
483                port: 9095,
484                ..Default::default()
485            }],
486            ..Default::default()
487        }
488    }
489
490    fn remove_voter_request(
491        cluster_id: Option<&str>,
492        voter_id: i32,
493        voter_directory_id: WireUuid,
494    ) -> RemoveRaftVoterRequest {
495        RemoveRaftVoterRequest {
496            cluster_id: cluster_id.map(str::to_owned),
497            voter_id,
498            voter_directory_id,
499            ..Default::default()
500        }
501    }
502
503    fn update_voter_request() -> UpdateRaftVoterRequest {
504        UpdateRaftVoterRequest {
505            // A request with no cluster id skips Kafka's cluster check.
506            cluster_id: None,
507            voter_id: 2,
508            voter_directory_id: DIRECTORY,
509            current_leader_epoch: 7,
510            k_raft_version_feature: KRaftVersionFeature {
511                min_supported_version: 0,
512                max_supported_version: 1,
513                ..Default::default()
514            },
515            listeners: vec![UpdateListener {
516                name: "CONTROLLER".into(),
517                host: "controller-2".into(),
518                port: 9094,
519                ..Default::default()
520            }],
521            ..Default::default()
522        }
523    }
524
525    fn test_cluster_uuid() -> uuid::Uuid {
526        uuid::Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10)
527    }
528
529    #[yare::parameterized(
530        hyphenated = { test_cluster_uuid().to_string(), true },
531        kafka_base64 = { URL_SAFE_NO_PAD.encode(test_cluster_uuid().as_bytes()), true },
532        another_cluster = { uuid::Uuid::from_u128(7).to_string(), false },
533        not_an_id = { "cluster-b".to_owned(), false },
534    )]
535    fn a_cluster_id_matches_in_either_form(request: String, names_it: bool) {
536        let local = test_cluster_uuid().to_string();
537        check!(
538            names_this_cluster(&request, &local) == names_it,
539            "{request}"
540        );
541    }
542
543    #[test]
544    fn replica_key_names_the_directory_in_kafka_form() {
545        check!(replica_key(3, WireUuid::ZERO) == "ReplicaKey(id=3, directoryId=<undefined>)");
546        check!(replica_key(3, DIRECTORY) == "ReplicaKey(id=3, directoryId=CQkJCQkJCQkJCQkJCQkJCQ)");
547    }
548
549    #[yare::parameterized(
550        foreign_cluster_follower = {
551            "foreign cluster id, on a follower",
552            AddRaftVoterRequest { cluster_id: Some("cluster-b".into()), ..add_voter_request() },
553            false,
554            Some(foreign_cluster_refusal())
555        },
556        follower = {
557            "a follower",
558            add_voter_request(),
559            false,
560            Some((NOT_LEADER_OR_FOLLOWER, None))
561        },
562        invalid_follower = {
563            "a follower, with an invalid voter",
564            AddRaftVoterRequest { voter_id: -1, ..add_voter_request() },
565            false,
566            Some((NOT_LEADER_OR_FOLLOWER, None))
567        },
568        negative_voter = {
569            "negative voter id",
570            AddRaftVoterRequest { voter_id: -1, ..add_voter_request() },
571            true,
572            Some((INVALID_REQUEST, Some("Add voter request didn't include a valid voter".into())))
573        },
574        zero_directory = {
575            "zero directory id",
576            AddRaftVoterRequest { voter_directory_id: WireUuid::ZERO, ..add_voter_request() },
577            true,
578            Some((INVALID_REQUEST, Some("Add voter request didn't include a valid voter".into())))
579        },
580        no_listeners = {
581            "no listeners",
582            AddRaftVoterRequest { listeners: Vec::new(), ..add_voter_request() },
583            true,
584            Some((INVALID_REQUEST, Some("Add voter request didn't include a valid voter".into())))
585        },
586        no_controller_listener = {
587            "no controller listener",
588            {
589                let mut request = add_voter_request();
590                request.listeners[0].name = "PLAINTEXT".into();
591                request
592            },
593            true,
594            Some((
595                INVALID_REQUEST,
596                Some(
597                    "Add voter request didn't include the endpoint for the default listener CONTROLLER"
598                        .into(),
599                ),
600            )),
601        },
602        this_cluster = {
603            "this cluster",
604            AddRaftVoterRequest { cluster_id: Some(CLUSTER.into()), ..add_voter_request() },
605            true,
606            None
607        },
608        no_cluster_id = { "no cluster id", add_voter_request(), true, None },
609    )]
610    fn add_voter_checks_run_in_kafka_order(
611        label: &'static str,
612        request: AddRaftVoterRequest,
613        is_leader: bool,
614        expected: Option<Refusal>,
615    ) {
616        check!(
617            add_voter_refusal(&request, CLUSTER, &quorum(is_leader)) == expected,
618            "{label}"
619        );
620    }
621
622    #[yare::parameterized(
623        foreign_cluster = {
624            "foreign cluster id",
625            remove_voter_request(Some("cluster-b"), 2, DIRECTORY),
626            true,
627            Some(foreign_cluster_refusal()),
628        },
629        follower = {
630            "a follower",
631            remove_voter_request(None, -1, DIRECTORY),
632            false,
633            Some((NOT_LEADER_OR_FOLLOWER, None)),
634        },
635        negative_voter = {
636            "negative voter id",
637            remove_voter_request(None, -1, DIRECTORY),
638            true,
639            Some((INVALID_REQUEST, Some("Remove voter request didn't include a valid voter".into()))),
640        },
641        zero_directory = {
642            "zero directory id",
643            remove_voter_request(None, 2, WireUuid::ZERO),
644            true,
645            Some((INVALID_REQUEST, Some("Remove voter request didn't include a valid voter".into()))),
646        },
647        valid_voter = {
648            "a valid voter",
649            remove_voter_request(Some(CLUSTER), 2, DIRECTORY),
650            true,
651            None,
652        },
653    )]
654    fn remove_voter_checks_run_in_kafka_order(
655        label: &'static str,
656        request: RemoveRaftVoterRequest,
657        is_leader: bool,
658        expected: Option<Refusal>,
659    ) {
660        check!(
661            remove_voter_refusal(&request, CLUSTER, &quorum(is_leader)) == expected,
662            "{label}"
663        );
664    }
665
666    #[yare::parameterized(
667        foreign_cluster = {
668            "foreign cluster id",
669            UpdateRaftVoterRequest { cluster_id: Some("cluster-b".into()), ..update_voter_request() },
670            true,
671            Some(INCONSISTENT_CLUSTER_ID),
672        },
673        epoch_below = {
674            "an epoch below the local one",
675            UpdateRaftVoterRequest { current_leader_epoch: 6, ..update_voter_request() },
676            true,
677            Some(FENCED_LEADER_EPOCH),
678        },
679        epoch_above = {
680            "an epoch above the local one",
681            UpdateRaftVoterRequest { current_leader_epoch: 8, ..update_voter_request() },
682            true,
683            Some(UNKNOWN_LEADER_EPOCH),
684        },
685        follower = { "a follower", update_voter_request(), false, Some(NOT_LEADER_OR_FOLLOWER) },
686        zero_directory = {
687            "zero directory id",
688            UpdateRaftVoterRequest { voter_directory_id: WireUuid::ZERO, ..update_voter_request() },
689            true,
690            Some(INVALID_REQUEST),
691        },
692        negative_kraft_version = {
693            "a negative kraft.version",
694            UpdateRaftVoterRequest {
695                k_raft_version_feature: KRaftVersionFeature {
696                    min_supported_version: -1,
697                    ..update_voter_request().k_raft_version_feature
698                },
699                ..update_voter_request()
700            },
701            true,
702            Some(INVALID_REQUEST),
703        },
704        inverted_kraft_version = {
705            "an inverted kraft.version range",
706            UpdateRaftVoterRequest {
707                k_raft_version_feature: KRaftVersionFeature {
708                    min_supported_version: 2,
709                    ..update_voter_request().k_raft_version_feature
710                },
711                ..update_voter_request()
712            },
713            true,
714            Some(INVALID_REQUEST),
715        },
716        no_controller_listener = {
717            "no controller listener",
718            UpdateRaftVoterRequest {
719                listeners: vec![UpdateListener {
720                    name: "PLAINTEXT".into(),
721                    ..Default::default()
722                }],
723                ..update_voter_request()
724            },
725            true,
726            Some(INVALID_REQUEST),
727        },
728        valid_update = { "a valid update", update_voter_request(), true, None },
729    )]
730    fn update_voter_checks_run_in_kafka_order(
731        label: &'static str,
732        request: UpdateRaftVoterRequest,
733        is_leader: bool,
734        expected: Option<i16>,
735    ) {
736        check!(
737            update_voter_refusal(&request, CLUSTER, &quorum(is_leader)) == expected,
738            "{label}"
739        );
740    }
741
742    #[yare::parameterized(
743        known_leader = {
744            "a known leader",
745            quorum(false),
746            CurrentLeader {
747                leader_id: 1,
748                leader_epoch: 7,
749                host: "controller-1".into(),
750                port: 9093,
751                ..Default::default()
752            },
753        },
754        no_leader = {
755            "no leader",
756            QuorumStateSnapshot { leader_id: None, ..quorum(false) },
757            CurrentLeader { leader_id: -1, leader_epoch: 7, ..Default::default() }
758        },
759    )]
760    fn update_voter_answers_name_the_leader(
761        label: &'static str,
762        quorum: QuorumStateSnapshot,
763        expected: CurrentLeader,
764    ) {
765        check!(update_voter_current_leader(&quorum) == expected, "{label}");
766    }
767
768    #[yare::parameterized(
769        committed = { "committed", Ok(ReconfigOutcome::Committed), (0, None) },
770        not_leader = {
771            "not the leader",
772            Ok(ReconfigOutcome::NotLeader { leader: Some(NodeId(2)) }),
773            (NOT_LEADER_OR_FOLLOWER, None)
774        },
775        lagging = {
776            "a lagging candidate",
777            Err(RaftError::VoterNotCaughtUp { id: NodeId(3), lag: 10 }),
778            (
779                REQUEST_TIMED_OUT,
780                Some(format!(
781                    "Aborted add voter operation for {TEST_VOTER_KEY} since it is lagging behind"
782                )),
783            ),
784        },
785        pending_change = {
786            "a pending change",
787            Err(RaftError::ReconfigInProgress),
788            (
789                REQUEST_TIMED_OUT,
790                Some("Request timed out waiting for leader to handle previous voter change request".into()),
791            ),
792        },
793        duplicate_voter = {
794            "a duplicate voter",
795            Err(RaftError::DuplicateVoter(NodeId(3))),
796            (
797                DUPLICATE_VOTER,
798                Some(format!(
799                    "The voter id for {TEST_VOTER_KEY} is already part of the set of voters"
800                )),
801            ),
802        },
803        unknown_voter = {
804            "an unknown voter",
805            Err(RaftError::VoterNotFound(NodeId(3))),
806            (VOTER_NOT_FOUND, Some(format!("Cannot remove voter {TEST_VOTER_KEY} from the set of voters")))
807        },
808        rejected_change = {
809            "a rejected change",
810            Err(RaftError::ReconfigRejected("empty voter set".into())),
811            (INVALID_REQUEST, Some("empty voter set".into()))
812        },
813    )]
814    fn reconfiguration_outcomes_get_kafka_codes(
815        label: &'static str,
816        result: Result<ReconfigOutcome, RaftError>,
817        expected: Refusal,
818    ) {
819        check!(
820            reconfiguration_refusal(result, VoterOperation::Add, 3, DIRECTORY) == expected,
821            "{label}"
822        );
823    }
824
825    /// `AddVoterHandler` and `RemoveVoterHandler` name their own operation in
826    /// the `UNSUPPORTED_VERSION` message, and the level is the finalized one.
827    #[yare::parameterized(
828        add_v0 = { VoterOperation::Add, 0, "adding" },
829        remove_v0 = { VoterOperation::Remove, 0, "removing" },
830        add_v2 = { VoterOperation::Add, 2, "adding" },
831    )]
832    fn unsupported_kraft_version_names_the_operation(
833        operation: VoterOperation,
834        level: u16,
835        verb: &'static str,
836    ) {
837        check!(
838            reconfiguration_refusal(
839                Err(RaftError::UnsupportedKraftVersion(level)),
840                operation,
841                3,
842                DIRECTORY
843            ) == (
844                UNSUPPORTED_VERSION,
845                Some(format!(
846                    "Cluster doesn't support {verb} voter because the kraft.version feature is {level}"
847                ))
848            ),
849            "{operation:?} at {level}"
850        );
851    }
852}
krabka_raft/voter_requests.rs

krabka_raft/
voter_requests.rs
