krabka_client_admin/
quorum.rs

1//! KIP-853 metadata-quorum administration: `DescribeQuorum`, `AddRaftVoter`
2//! and `RemoveRaftVoter`, as Kafka's `Admin.describeMetadataQuorum`,
3//! `addRaftVoter` and `removeRaftVoter`.
4
5use std::collections::BTreeSet;
6
7use krabka_protocol::{
8    owned::{
9        add_raft_voter_request::{AddRaftVoterRequest, Listener},
10        common::describe_quorum_response::replica_state::ReplicaState,
11        describe_quorum_request::{
12            DescribeQuorumRequest, PartitionData as RequestPartition, TopicData as RequestTopic,
13        },
14        describe_quorum_response,
15        remove_raft_voter_request::RemoveRaftVoterRequest,
16    },
17    primitives::uuid::Uuid as ProtoUuid,
18};
19
20use crate::{AdminClient, AdminError, NOT_CONTROLLER, kafka_error_name, retry::ControllerRetry};
21
22const CLUSTER_METADATA_TOPIC: &str = "__cluster_metadata";
23
24/// One listener endpoint of a voter, as Kafka's `RaftVoterEndpoint`.
25#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
26pub struct RaftVoterEndpoint {
27    listener: String,
28    host: String,
29    port: u16,
30}
31
32impl RaftVoterEndpoint {
33    /// An endpoint of `listener` at `host:port`.
34    ///
35    /// # Errors
36    /// Returns [`AdminError::InvalidArgument`] when `listener` is empty, has
37    /// leading or trailing whitespace, or is not in upper case, as Kafka's
38    /// `RaftVoterEndpoint` constructor refuses it.
39    pub fn new(
40        listener: impl Into<String>,
41        host: impl Into<String>,
42        port: u16,
43    ) -> Result<Self, AdminError> {
44        let listener = listener.into();
45        if listener.trim() != listener {
46            return Err(AdminError::InvalidArgument(format!(
47                "listener {listener:?}: leading or trailing whitespace is not allowed"
48            )));
49        }
50        if listener.is_empty() {
51            return Err(AdminError::InvalidArgument(
52                "listener: empty string is not allowed".to_owned(),
53            ));
54        }
55        if listener.to_uppercase() != listener {
56            return Err(AdminError::InvalidArgument(format!(
57                "listener {listener:?}: string must be UPPERCASE"
58            )));
59        }
60        Ok(Self {
61            listener,
62            host: host.into(),
63            port,
64        })
65    }
66
67    /// The listener name.
68    #[must_use]
69    pub fn listener(&self) -> &str {
70        &self.listener
71    }
72
73    /// The advertised host.
74    #[must_use]
75    pub fn host(&self) -> &str {
76        &self.host
77    }
78
79    /// The advertised port.
80    #[must_use]
81    pub const fn port(&self) -> u16 {
82        self.port
83    }
84}
85
86/// The `AddRaftVoter` request of one voter, as Kafka's `addRaftVoter`
87/// `createRequest` builds it. Each endpoint appears once, as Kafka takes a
88/// `Set`. `ack_when_committed` keeps its schema default, `true`, which
89/// Kafka's client does not change.
90fn add_raft_voter_request(
91    cluster_id: Option<&str>,
92    voter_id: i32,
93    voter_directory_id: uuid::Uuid,
94    endpoints: &[RaftVoterEndpoint],
95    timeout_ms: i32,
96) -> AddRaftVoterRequest {
97    let mut seen = BTreeSet::new();
98    AddRaftVoterRequest {
99        cluster_id: cluster_id.map(str::to_owned),
100        timeout_ms,
101        voter_id,
102        voter_directory_id: ProtoUuid(*voter_directory_id.as_bytes()),
103        listeners: endpoints
104            .iter()
105            .filter(|endpoint| seen.insert(*endpoint))
106            .map(|endpoint| Listener {
107                name: endpoint.listener.clone(),
108                host: endpoint.host.clone(),
109                port: endpoint.port,
110                ..Default::default()
111            })
112            .collect(),
113        ..Default::default()
114    }
115}
116
117/// The `RemoveRaftVoter` request of one voter, as Kafka's `removeRaftVoter`
118/// `createRequest` builds it: the cluster ID as the caller gave it, or null.
119fn remove_raft_voter_request(
120    cluster_id: Option<&str>,
121    voter_id: i32,
122    voter_directory_id: uuid::Uuid,
123) -> RemoveRaftVoterRequest {
124    RemoveRaftVoterRequest {
125        cluster_id: cluster_id.map(str::to_owned),
126        voter_id,
127        voter_directory_id: ProtoUuid(*voter_directory_id.as_bytes()),
128        ..Default::default()
129    }
130}
131
132/// `Ok` for code 0, and the broker error of `api` for any other code.
133fn voter_result(api: &'static str, code: i16, message: Option<String>) -> Result<(), AdminError> {
134    if code == 0 {
135        Ok(())
136    } else {
137        Err(broker_error(api, code, message))
138    }
139}
140
141#[derive(Debug, Clone, PartialEq, Eq)]
142pub struct QuorumReplica {
143    pub node_id: i32,
144    pub directory_id: uuid::Uuid,
145    pub log_end_offset: i64,
146    pub last_fetch_timestamp: i64,
147    pub last_caught_up_timestamp: i64,
148}
149
150#[derive(Debug, Clone, Default, PartialEq, Eq)]
151pub struct MetadataQuorum {
152    pub leader_id: i32,
153    pub leader_epoch: i32,
154    pub high_watermark: i64,
155    pub voters: Vec<QuorumReplica>,
156    pub observers: Vec<QuorumReplica>,
157    /// The listener endpoints of each node, as Kafka's `QuorumInfo.nodes()`
158    /// reads them from the response's `Nodes` (`DescribeQuorum` v2 and
159    /// later). Empty when the controller answered at v0 or v1.
160    pub nodes: Vec<QuorumNode>,
161}
162
163/// One node of `QuorumInfo.nodes()`: its id and listener endpoints.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct QuorumNode {
166    pub node_id: i32,
167    pub endpoints: Vec<RaftVoterEndpoint>,
168}
169
170fn broker_error(api: &'static str, code: i16, message: Option<String>) -> AdminError {
171    AdminError::Broker {
172        api,
173        code,
174        name: kafka_error_name(code),
175        message,
176    }
177}
178
179fn replica(value: &ReplicaState) -> QuorumReplica {
180    QuorumReplica {
181        node_id: value.replica_id,
182        directory_id: uuid::Uuid::from_bytes(value.replica_directory_id.0),
183        log_end_offset: value.log_end_offset,
184        last_fetch_timestamp: value.last_fetch_timestamp,
185        last_caught_up_timestamp: value.last_caught_up_timestamp,
186    }
187}
188
189/// A response node as Kafka's `DescribeQuorumResponse.getNodes` maps it. The
190/// listener fields come straight from the wire, so they are not re-validated
191/// the way a caller-built `RaftVoterEndpoint` is.
192fn quorum_node(node: &describe_quorum_response::Node) -> QuorumNode {
193    QuorumNode {
194        node_id: node.node_id,
195        endpoints: node
196            .listeners
197            .iter()
198            .map(|listener| RaftVoterEndpoint {
199                listener: listener.name.clone(),
200                host: listener.host.clone(),
201                port: listener.port,
202            })
203            .collect(),
204    }
205}
206
207impl AdminClient {
208    /// The call behind [`AdminClientLike::describe_metadata_quorum`](crate::AdminClientLike::describe_metadata_quorum).
209    pub(crate) async fn describe_metadata_quorum_impl(&self) -> Result<MetadataQuorum, AdminError> {
210        let response = self
211            .conn
212            .send(DescribeQuorumRequest {
213                topics: vec![RequestTopic {
214                    topic_name: CLUSTER_METADATA_TOPIC.into(),
215                    partitions: vec![RequestPartition {
216                        partition_index: 0,
217                        ..Default::default()
218                    }],
219                    ..Default::default()
220                }],
221                ..Default::default()
222            })
223            .await?;
224        if response.error_code != 0 {
225            return Err(broker_error(
226                "DescribeQuorum",
227                response.error_code,
228                response.error_message,
229            ));
230        }
231        let nodes = response.nodes.iter().map(quorum_node).collect();
232        let partition = response
233            .topics
234            .into_iter()
235            .find(|topic| topic.topic_name == CLUSTER_METADATA_TOPIC)
236            .and_then(|topic| {
237                topic
238                    .partitions
239                    .into_iter()
240                    .find(|partition| partition.partition_index == 0)
241            })
242            .ok_or_else(|| {
243                AdminError::Protocol("DescribeQuorum omitted __cluster_metadata partition 0".into())
244            })?;
245        if partition.error_code != 0 {
246            return Err(broker_error(
247                "DescribeQuorum",
248                partition.error_code,
249                partition.error_message,
250            ));
251        }
252        Ok(MetadataQuorum {
253            leader_id: partition.leader_id,
254            leader_epoch: partition.leader_epoch,
255            high_watermark: partition.high_watermark,
256            voters: partition.current_voters.iter().map(replica).collect(),
257            observers: partition.observers.iter().map(replica).collect(),
258            nodes,
259        })
260    }
261
262    /// Adds a voter to the metadata quorum (KIP-853), as Kafka's
263    /// `addRaftVoter` operation does.
264    ///
265    /// The client sends `AddRaftVoter` on its connection: the active
266    /// controller through controller bootstrap (KIP-919), or a broker, which
267    /// forwards it, as Kafka's `LeastLoadedBrokerOrActiveKController` picks.
268    /// `cluster_id` is Kafka's `AddRaftVoterOptions.clusterId`; `None` sends
269    /// a null cluster ID, which the controller does not check. The request
270    /// carries the time left before the call deadline as its `timeout_ms`.
271    ///
272    /// `NOT_CONTROLLER` (41) makes the client find the controller again and
273    /// resend the request with the backoff until `default.api.timeout.ms`
274    /// (60 s), as Kafka's `handleNotControllerError` does.
275    ///
276    /// # Errors
277    /// Returns [`AdminError::Broker`] with the controller's error, such as
278    /// `DUPLICATE_VOTER` (126), and with `REQUEST_TIMED_OUT` (7) at the
279    /// deadline, or a transport error.
280    pub async fn add_raft_voter(
281        &mut self,
282        cluster_id: Option<&str>,
283        voter_id: i32,
284        voter_directory_id: uuid::Uuid,
285        endpoints: &[RaftVoterEndpoint],
286    ) -> Result<(), AdminError> {
287        let mut retry = ControllerRetry::new("AddRaftVoter", self.retry);
288        loop {
289            let request = add_raft_voter_request(
290                cluster_id,
291                voter_id,
292                voter_directory_id,
293                endpoints,
294                retry.remaining_millis(),
295            );
296            let response = retry.bounded(self.conn.send(request)).await?;
297            if response.error_code != NOT_CONTROLLER {
298                return voter_result("AddRaftVoter", response.error_code, response.error_message);
299            }
300            retry.after_not_controller(self).await?;
301        }
302    }
303
304    /// The call behind [`AdminClientLike::remove_raft_voter`](crate::AdminClientLike::remove_raft_voter).
305    pub(crate) async fn remove_raft_voter_impl(
306        &self,
307        cluster_id: Option<&str>,
308        node_id: i32,
309        directory_id: uuid::Uuid,
310    ) -> Result<(), AdminError> {
311        let request = remove_raft_voter_request(cluster_id, node_id, directory_id);
312        let mut retry = ControllerRetry::new("RemoveRaftVoter", self.retry);
313        loop {
314            let response = retry.bounded(self.conn.send(request.clone())).await?;
315            if response.error_code != NOT_CONTROLLER {
316                return voter_result(
317                    "RemoveRaftVoter",
318                    response.error_code,
319                    response.error_message,
320                );
321            }
322            retry.after_not_controller(self).await?;
323        }
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    use std::sync::{Arc, Mutex};
330
331    use krabka_client_core::MockReply;
332    use krabka_protocol::owned::{
333        add_raft_voter_request, add_raft_voter_response::AddRaftVoterResponse,
334        remove_raft_voter_request, remove_raft_voter_response::RemoveRaftVoterResponse,
335    };
336
337    use super::*;
338    use crate::{
339        AdminClientLike as _,
340        partition_leaders::test_support::{
341            ApiVersionRange, decode_request, encode_response, fast_admin, scripted_broker,
342        },
343    };
344
345    const DIRECTORY: uuid::Uuid = uuid::Uuid::from_u128(0x0123_4567_89ab_cdef_0123_4567_89ab_cdef);
346
347    fn endpoint(listener: &str, host: &str, port: u16) -> RaftVoterEndpoint {
348        RaftVoterEndpoint::new(listener, host, port).expect("valid endpoint")
349    }
350
351    #[test]
352    fn endpoint_listener_must_be_trimmed_non_empty_upper_case() {
353        for (name, listener, valid) in [
354            ("upper case", "CONTROLLER", true),
355            ("upper case with digits and symbols", "CONTROLLER_2-X", true),
356            ("empty", "", false),
357            ("lower case", "controller", false),
358            ("leading whitespace", " CONTROLLER", false),
359            ("trailing whitespace", "CONTROLLER ", false),
360        ] {
361            let result = RaftVoterEndpoint::new(listener, "localhost", 9093);
362            assert2::assert!(
363                matches!(result, Err(AdminError::InvalidArgument(_))) != valid,
364                "case {name}"
365            );
366        }
367    }
368
369    #[test]
370    fn add_request_names_each_endpoint_once() {
371        let endpoints = [
372            endpoint("CONTROLLER", "c1", 9093),
373            endpoint("CONTROLLER", "c1", 9093),
374            endpoint("SSL", "c1", 9094),
375        ];
376        for (name, cluster_id, expected_cluster) in [
377            (
378                "with cluster id",
379                Some("MkU3OEVBNTcwNTJENDM2Qk"),
380                Some("MkU3OEVBNTcwNTJENDM2Qk"),
381            ),
382            ("without cluster id", None, None),
383        ] {
384            assert2::assert!(
385                add_raft_voter_request(cluster_id, 3, DIRECTORY, &endpoints, 1_000)
386                    == AddRaftVoterRequest {
387                        cluster_id: expected_cluster.map(str::to_owned),
388                        timeout_ms: 1_000,
389                        voter_id: 3,
390                        voter_directory_id: ProtoUuid(*DIRECTORY.as_bytes()),
391                        listeners: vec![
392                            Listener {
393                                name: "CONTROLLER".to_owned(),
394                                host: "c1".to_owned(),
395                                port: 9093,
396                                ..Default::default()
397                            },
398                            Listener {
399                                name: "SSL".to_owned(),
400                                host: "c1".to_owned(),
401                                port: 9094,
402                                ..Default::default()
403                            },
404                        ],
405                        ack_when_committed: true,
406                        ..Default::default()
407                    },
408                "case {name}"
409            );
410        }
411    }
412
413    #[test]
414    fn remove_request_sends_the_cluster_id_as_given() {
415        for (name, cluster_id) in [
416            ("with cluster id", Some("MkU3OEVBNTcwNTJENDM2Qk")),
417            ("without cluster id", None),
418        ] {
419            assert2::assert!(
420                remove_raft_voter_request(cluster_id, 3, DIRECTORY)
421                    == RemoveRaftVoterRequest {
422                        cluster_id: cluster_id.map(str::to_owned),
423                        voter_id: 3,
424                        voter_directory_id: ProtoUuid(*DIRECTORY.as_bytes()),
425                        ..Default::default()
426                    },
427                "case {name}"
428            );
429        }
430    }
431
432    /// `describe_metadata_quorum` reads the leader, voters, observers and,
433    /// from v2, every node's listener endpoints, as Kafka's
434    /// `DescribeQuorumResponse.getQuorumInfo` does. A v1 answer carries no
435    /// nodes.
436    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
437    async fn describe_metadata_quorum_reads_nodes_from_v2() {
438        use krabka_protocol::owned::{
439            common::describe_quorum_response::replica_state::ReplicaState,
440            describe_quorum_request,
441            describe_quorum_response::{
442                DescribeQuorumResponse, Listener, Node, PartitionData, TopicData,
443            },
444        };
445
446        let voter = |replica_id| ReplicaState {
447            replica_id,
448            log_end_offset: 40,
449            ..Default::default()
450        };
451        let response = DescribeQuorumResponse {
452            topics: vec![TopicData {
453                topic_name: CLUSTER_METADATA_TOPIC.into(),
454                partitions: vec![PartitionData {
455                    leader_id: 1,
456                    leader_epoch: 3,
457                    high_watermark: 41,
458                    current_voters: vec![voter(1), voter(2)],
459                    observers: vec![voter(7)],
460                    ..Default::default()
461                }],
462                ..Default::default()
463            }],
464            nodes: vec![Node {
465                node_id: 1,
466                listeners: vec![Listener {
467                    name: "CONTROLLER".into(),
468                    host: "c1.example".into(),
469                    port: 9093,
470                    ..Default::default()
471                }],
472                ..Default::default()
473            }],
474            ..Default::default()
475        };
476        let quorum = |nodes| MetadataQuorum {
477            leader_id: 1,
478            leader_epoch: 3,
479            high_watermark: 41,
480            voters: vec![
481                QuorumReplica {
482                    node_id: 1,
483                    directory_id: uuid::Uuid::nil(),
484                    log_end_offset: 40,
485                    last_fetch_timestamp: -1,
486                    last_caught_up_timestamp: -1,
487                },
488                QuorumReplica {
489                    node_id: 2,
490                    directory_id: uuid::Uuid::nil(),
491                    log_end_offset: 40,
492                    last_fetch_timestamp: -1,
493                    last_caught_up_timestamp: -1,
494                },
495            ],
496            observers: vec![QuorumReplica {
497                node_id: 7,
498                directory_id: uuid::Uuid::nil(),
499                log_end_offset: 40,
500                last_fetch_timestamp: -1,
501                last_caught_up_timestamp: -1,
502            }],
503            nodes,
504        };
505        for (name, max_version, expected) in [
506            (
507                "v2 carries nodes",
508                2,
509                quorum(vec![QuorumNode {
510                    node_id: 1,
511                    endpoints: vec![endpoint("CONTROLLER", "c1.example", 9093)],
512                }]),
513            ),
514            ("v1 carries none", 1, quorum(vec![])),
515        ] {
516            let answer = response.clone();
517            let broker = scripted_broker(
518                vec![
519                    ApiVersionRange::builder()
520                        .api_key(describe_quorum_request::API_KEY)
521                        .max_version(max_version)
522                        .build(),
523                ],
524                move |key, version, _, _| {
525                    if key == describe_quorum_request::API_KEY {
526                        MockReply::Respond(encode_response(&answer, version, true))
527                    } else {
528                        MockReply::Silent
529                    }
530                },
531            )
532            .await;
533            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
534
535            let described = admin.describe_metadata_quorum().await.expect("describes");
536
537            broker.stop();
538            assert2::assert!(described == expected, "case {name}");
539        }
540    }
541
542    /// Kafka's `addRaftVoter` and `removeRaftVoter` resend after
543    /// `NOT_CONTROLLER` and fail at once on every other error code.
544    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
545    async fn voter_calls_retry_not_controller_and_stop_on_other_codes() {
546        for (name, add, codes, expected) in [
547            ("add succeeds", true, vec![0], (Ok(()), 1)),
548            (
549                "add after not controller",
550                true,
551                vec![NOT_CONTROLLER, 0],
552                (Ok(()), 2),
553            ),
554            (
555                "add duplicate voter is final",
556                true,
557                vec![126],
558                (Err(("AddRaftVoter", 126, Some("dup".to_owned()))), 1),
559            ),
560            ("remove succeeds", false, vec![0], (Ok(()), 1)),
561            (
562                "remove after not controller",
563                false,
564                vec![NOT_CONTROLLER, NOT_CONTROLLER, 0],
565                (Ok(()), 3),
566            ),
567            (
568                "remove voter not found is final",
569                false,
570                vec![127],
571                (Err(("RemoveRaftVoter", 127, Some("dup".to_owned()))), 1),
572            ),
573        ] {
574            let requests = Arc::new(Mutex::new(Vec::<(i16, i32)>::new()));
575            let handler_requests = Arc::clone(&requests);
576            let broker = scripted_broker(
577                vec![
578                    ApiVersionRange::builder()
579                        .api_key(add_raft_voter_request::API_KEY)
580                        .max_version(1)
581                        .build(),
582                    ApiVersionRange::builder()
583                        .api_key(remove_raft_voter_request::API_KEY)
584                        .build(),
585                ],
586                move |api_key, version, body, _| {
587                    let mut requests = handler_requests.lock().expect("requests lock");
588                    let code = codes[requests.len().min(codes.len() - 1)];
589                    let message = (code != 0).then(|| "dup".to_owned());
590                    let reply = match api_key {
591                        add_raft_voter_request::API_KEY => {
592                            let request: AddRaftVoterRequest = decode_request(body, version, true);
593                            requests.push((version, request.voter_id));
594                            encode_response(
595                                &AddRaftVoterResponse {
596                                    error_code: code,
597                                    error_message: message,
598                                    ..Default::default()
599                                },
600                                version,
601                                true,
602                            )
603                        }
604                        remove_raft_voter_request::API_KEY => {
605                            let request: RemoveRaftVoterRequest =
606                                decode_request(body, version, true);
607                            requests.push((version, request.voter_id));
608                            encode_response(
609                                &RemoveRaftVoterResponse {
610                                    error_code: code,
611                                    error_message: message,
612                                    ..Default::default()
613                                },
614                                version,
615                                true,
616                            )
617                        }
618                        _ => return MockReply::Silent,
619                    };
620                    drop(requests);
621                    MockReply::Respond(reply)
622                },
623            )
624            .await;
625            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
626
627            let result = if add {
628                admin
629                    .add_raft_voter(None, 3, DIRECTORY, &[endpoint("CONTROLLER", "c3", 9093)])
630                    .await
631            } else {
632                admin.remove_raft_voter(None, 3, DIRECTORY).await
633            }
634            .map_err(|error| match error {
635                AdminError::Broker {
636                    api, code, message, ..
637                } => (api, code, message),
638                other => panic!("case {name}: unexpected error {other:?}"),
639            });
640
641            broker.stop();
642            let version = i16::from(add);
643            assert2::assert!(
644                (result, requests.lock().expect("requests lock").clone())
645                    == (expected.0, vec![(version, 3); expected.1]),
646                "case {name}"
647            );
648        }
649    }
650}
krabka_client_admin/quorum.rs

krabka_client_admin/
quorum.rs
