krabka_client_admin/
features.rs

1//! Cluster feature administration, as Kafka's `Admin.describeFeatures` and
2//! `Admin.updateFeatures`.
3
4use std::{collections::BTreeMap, time::Duration};
5
6use krabka_protocol::owned::{
7    api_versions_request::ApiVersionsRequest,
8    api_versions_response::ApiVersionsResponse,
9    update_features_request::{FeatureUpdateKey, UpdateFeaturesRequest},
10    update_features_response::UpdateFeaturesResponse,
11};
12use krabka_units::{Time, convert::TimeExt as _};
13
14use crate::{
15    AdminClient, AdminError, KafkaError, MetadataVersionUpdate, NOT_CONTROLLER, kafka_error_if,
16    kafka_error_name,
17    log_dirs::{NodeTarget, call_node},
18    retry::{ControllerRetry, CoordinatorRetry, RetryPolicy},
19};
20
21const METADATA_VERSION_FEATURE: &str = "metadata.version";
22
23/// `UNKNOWN_SERVER_ERROR`: Kafka's code for a plain `ApiException`.
24const UNKNOWN_SERVER_ERROR: i16 = -1;
25
26/// The lowest `ApiVersions` version whose response carries features.
27const FEATURES_API_VERSIONS_MIN_VERSION: i16 = 3;
28
29/// The lowest `UpdateFeatures` version with `UpgradeType` and `ValidateOnly`.
30/// Kafka's generated `UpdateFeaturesRequestData` refuses to write either
31/// field at a non-default value at v0, where neither exists.
32const UPGRADE_TYPE_MIN_VERSION: i16 = 1;
33
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct FeatureRange {
36    pub name: String,
37    pub min_version: i16,
38    pub max_version: i16,
39}
40
41/// The features that one node reports, as Kafka's `FeatureMetadata`.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct FeatureMetadata {
44    pub supported: Vec<FeatureRange>,
45    pub finalized: Vec<FeatureRange>,
46    /// The epoch of the finalized features. `None` when the node reports a
47    /// negative epoch, as Kafka's `finalizedFeaturesEpoch` is empty then.
48    pub finalized_features_epoch: Option<i64>,
49}
50
51/// What kind of change a [`FeatureUpdate`] asks for, as Kafka's
52/// `FeatureUpdate.UpgradeType`.
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
54pub enum UpgradeType {
55    /// Raise the finalized level only.
56    Upgrade,
57    /// Lower the finalized level, or delete the feature, only where no
58    /// metadata is lost.
59    SafeDowngrade,
60    /// Lower the finalized level, or delete the feature, even where metadata
61    /// is lost.
62    UnsafeDowngrade,
63}
64
65impl UpgradeType {
66    /// The `UpgradeType` code of the wire request: 1, 2 or 3.
67    #[must_use]
68    pub const fn code(self) -> i8 {
69        match self {
70            Self::Upgrade => 1,
71            Self::SafeDowngrade => 2,
72            Self::UnsafeDowngrade => 3,
73        }
74    }
75}
76
77/// One finalized-feature change, as Kafka's `FeatureUpdate`. The feature name
78/// is the key of the map that [`AdminClient::update_features`] takes.
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub struct FeatureUpdate {
81    max_version_level: i16,
82    upgrade_type: UpgradeType,
83}
84
85impl FeatureUpdate {
86    /// A change of the feature to `max_version_level`. Level 0 deletes the
87    /// feature and needs a downgrade type.
88    ///
89    /// # Errors
90    /// Returns [`AdminError::InvalidArgument`] for level 0 with
91    /// [`UpgradeType::Upgrade`] and for a negative level, as Kafka's
92    /// constructor throws `IllegalArgumentException`.
93    pub fn new(max_version_level: i16, upgrade_type: UpgradeType) -> Result<Self, AdminError> {
94        if max_version_level == 0 && upgrade_type == UpgradeType::Upgrade {
95            return Err(AdminError::InvalidArgument(format!(
96                "The upgradeType flag should be set to SAFE_DOWNGRADE or UNSAFE_DOWNGRADE when \
97                 the provided maxVersionLevel:{max_version_level} is < 1."
98            )));
99        }
100        if max_version_level < 0 {
101            return Err(AdminError::InvalidArgument(
102                "Cannot specify a negative version level.".into(),
103            ));
104        }
105        Ok(Self {
106            max_version_level,
107            upgrade_type,
108        })
109    }
110
111    /// The new maximum finalized level.
112    #[must_use]
113    pub const fn max_version_level(&self) -> i16 {
114        self.max_version_level
115    }
116
117    /// The kind of change.
118    #[must_use]
119    pub const fn upgrade_type(&self) -> UpgradeType {
120        self.upgrade_type
121    }
122}
123
124/// Options of [`AdminClient::update_features`], as Kafka's
125/// `UpdateFeaturesOptions`.
126#[derive(Debug, Clone, Copy, Default, PartialEq, bon::Builder)]
127pub struct UpdateFeaturesOptions {
128    /// The deadline of the call, including retries. Each request carries the
129    /// time that remains as its `timeout_ms`. `None` uses the client's
130    /// `default.api.timeout.ms`.
131    #[builder(with = |millis: i64| Time::from_millis(millis))]
132    pub timeout: Option<Time>,
133    /// Ask the controller to validate the updates and apply none. Needs
134    /// `UpdateFeatures` v1.
135    #[builder(default)]
136    pub validate_only: bool,
137}
138
139/// Options of [`AdminClient::describe_features`], as Kafka's
140/// `DescribeFeaturesOptions`.
141#[derive(Debug, Clone, Copy, Default, PartialEq, bon::Builder)]
142pub struct DescribeFeaturesOptions {
143    /// The deadline of the call, including retries. `None` uses the client's
144    /// `default.api.timeout.ms`.
145    #[builder(with = |millis: i64| Time::from_millis(millis))]
146    pub timeout: Option<Time>,
147    /// The node to ask. `None` asks the node of the client's connection, as
148    /// Kafka's `LeastLoadedBrokerOrActiveKController` picks one. A node id
149    /// goes to that broker, or to that controller with a controller
150    /// bootstrap, as Kafka's `ConstantNodeIdProvider` does.
151    pub node_id: Option<i32>,
152}
153
154/// The result of each feature of an [`AdminClient::update_features`] call,
155/// as Kafka's `UpdateFeaturesResult.values`.
156pub type UpdateFeaturesResults = BTreeMap<String, Result<(), KafkaError>>;
157
158/// `base` with the call deadline `timeout`, when set.
159fn call_policy(timeout: Option<Time>, base: RetryPolicy) -> RetryPolicy {
160    timeout.map_or(base, |timeout| RetryPolicy {
161        timeout: Duration::from_millis(u64::try_from(timeout.millis_i64()).unwrap_or(0)),
162        ..base
163    })
164}
165
166/// The `UpdateFeatures` request of `updates`, as `KafkaAdminClient`'s
167/// `updateFeatures` builds it. It sets `UpgradeType` and never the v0
168/// `AllowDowngrade`, which stays `false`.
169fn update_request(
170    updates: &BTreeMap<String, FeatureUpdate>,
171    options: UpdateFeaturesOptions,
172    timeout_ms: i32,
173) -> UpdateFeaturesRequest {
174    UpdateFeaturesRequest {
175        timeout_ms,
176        feature_updates: updates
177            .iter()
178            .map(|(feature, update)| FeatureUpdateKey {
179                feature: feature.clone(),
180                max_version_level: update.max_version_level,
181                upgrade_type: update.upgrade_type.code(),
182                ..Default::default()
183            })
184            .collect(),
185        validate_only: options.validate_only,
186        ..Default::default()
187    }
188}
189
190/// The lowest `UpdateFeatures` version that can carry `updates` and
191/// `options`. Kafka's v0 serializer throws `UnsupportedVersionException` for
192/// a non-default `UpgradeType` or `ValidateOnly`, so a downgrade or a
193/// validate-only call needs v1.
194fn update_min_version(
195    updates: &BTreeMap<String, FeatureUpdate>,
196    options: UpdateFeaturesOptions,
197) -> i16 {
198    let needs_v1 = options.validate_only
199        || updates
200            .values()
201            .any(|update| update.upgrade_type != UpgradeType::Upgrade);
202    if needs_v1 {
203        UPGRADE_TYPE_MIN_VERSION
204    } else {
205        0
206    }
207}
208
209impl AdminClient {
210    /// Returns the supported and cluster-finalized feature ranges of one node,
211    /// as Kafka's `describeFeatures`.
212    ///
213    /// # Errors
214    /// Returns a transport, protocol, or broker error. With a node id, a node
215    /// that the metadata does not name, or that does not answer, is retried
216    /// until the deadline and then gives `REQUEST_TIMED_OUT` (7).
217    pub async fn describe_features(
218        &self,
219        options: DescribeFeaturesOptions,
220    ) -> Result<FeatureMetadata, AdminError> {
221        let request = ApiVersionsRequest {
222            client_software_name: "krabka-client-rs".into(),
223            client_software_version: env!("CARGO_PKG_VERSION").into(),
224            ..Default::default()
225        };
226        let policy = call_policy(options.timeout, self.retry);
227        let response = if let Some(node_id) = options.node_id {
228            call_node(
229                &self.conn,
230                NodeTarget {
231                    node_id,
232                    supports_controllers: true,
233                    min_version: Some(FEATURES_API_VERSIONS_MIN_VERSION),
234                },
235                &self.options,
236                request,
237                CoordinatorRetry::new(policy),
238            )
239            .await
240            .map_err(|error| AdminError::Broker {
241                api: "ApiVersions",
242                code: error.code,
243                name: error.name,
244                message: error.message,
245            })?
246        } else {
247            ControllerRetry::new("ApiVersions", policy)
248                .bounded(
249                    self.conn
250                        .send_at_least(request, FEATURES_API_VERSIONS_MIN_VERSION),
251                )
252                .await?
253        };
254        parse_feature_metadata(response)
255    }
256
257    /// Applies finalized-feature updates through `UpdateFeatures` on the
258    /// active controller, as Kafka's `updateFeatures`.
259    ///
260    /// Every feature of `updates` gets a result. A top-level error of the
261    /// controller fails every feature with it. A v2 response with no
262    /// per-feature results and no error succeeds every feature, and a
263    /// feature that a v0/v1 response leaves out fails with
264    /// `UNKNOWN_SERVER_ERROR`, as Kafka does.
265    ///
266    /// # Errors
267    /// Returns [`AdminError::InvalidArgument`], and sends nothing, for no
268    /// updates or a blank feature name, as Kafka throws
269    /// `IllegalArgumentException`. Returns
270    /// [`krabka_client_core::ClientError::IncompatibleVersion`] before
271    /// sending when a downgrade or [`UpdateFeaturesOptions::validate_only`]
272    /// meets a controller with only v0, as Kafka's request serializer throws
273    /// `UnsupportedVersionException`. A `NOT_CONTROLLER` answer is retried
274    /// after controller discovery until the deadline, then gives
275    /// `REQUEST_TIMED_OUT` (7). Transport errors are returned as they are.
276    pub async fn update_features(
277        &mut self,
278        updates: &BTreeMap<String, FeatureUpdate>,
279        options: UpdateFeaturesOptions,
280    ) -> Result<UpdateFeaturesResults, AdminError> {
281        if updates.is_empty() {
282            return Err(AdminError::InvalidArgument(
283                "Feature updates can not be null or empty.".into(),
284            ));
285        }
286        if updates.keys().any(|feature| feature.trim().is_empty()) {
287            return Err(AdminError::InvalidArgument(
288                "Provided feature can not be empty.".into(),
289            ));
290        }
291        let min_version = update_min_version(updates, options);
292        let mut retry =
293            ControllerRetry::new("UpdateFeatures", call_policy(options.timeout, self.retry));
294        loop {
295            let request = update_request(updates, options, retry.remaining_millis());
296            let response = retry
297                .bounded(self.conn.send_at_least(request, min_version))
298                .await?;
299            if response.error_code != NOT_CONTROLLER {
300                return Ok(update_results(updates, response));
301            }
302            retry.after_not_controller(self).await?;
303        }
304    }
305
306    /// The call behind [`AdminClientLike::update_metadata_version`](crate::AdminClientLike::update_metadata_version).
307    pub(crate) async fn update_metadata_version_impl(
308        &mut self,
309        level: i16,
310        upgrade_type: UpgradeType,
311        timeout: Time,
312    ) -> Result<MetadataVersionUpdate, AdminError> {
313        let updates = maplit::btreemap! {
314            METADATA_VERSION_FEATURE.to_owned() => FeatureUpdate::new(level, upgrade_type)?,
315        };
316        let results = self
317            .update_features(
318                &updates,
319                UpdateFeaturesOptions::builder()
320                    .timeout(timeout.millis_i64())
321                    .validate_only(false)
322                    .build(),
323            )
324            .await?;
325        if let Some(Err(error)) = results.into_values().next() {
326            return Err(AdminError::Broker {
327                api: "UpdateFeatures",
328                code: error.code,
329                name: error.name,
330                message: error.message,
331            });
332        }
333        Ok(MetadataVersionUpdate { level })
334    }
335}
336
337fn parse_feature_metadata(response: ApiVersionsResponse) -> Result<FeatureMetadata, AdminError> {
338    if response.error_code != 0 {
339        return Err(AdminError::Broker {
340            api: "ApiVersions",
341            code: response.error_code,
342            name: kafka_error_name(response.error_code),
343            message: None,
344        });
345    }
346    Ok(FeatureMetadata {
347        supported: response
348            .supported_features
349            .into_iter()
350            .map(|feature| FeatureRange {
351                name: feature.name,
352                min_version: feature.min_version,
353                max_version: feature.max_version,
354            })
355            .collect(),
356        finalized: response
357            .finalized_features
358            .into_iter()
359            .map(|feature| FeatureRange {
360                name: feature.name,
361                min_version: feature.min_version_level,
362                max_version: feature.max_version_level,
363            })
364            .collect(),
365        finalized_features_epoch: (response.finalized_features_epoch >= 0)
366            .then_some(response.finalized_features_epoch),
367    })
368}
369
370/// The result of each feature of `updates` from `response`, as
371/// `KafkaAdminClient`'s `updateFeatures` completes its futures.
372fn update_results(
373    updates: &BTreeMap<String, FeatureUpdate>,
374    response: UpdateFeaturesResponse,
375) -> UpdateFeaturesResults {
376    if let Some(error) = kafka_error_if(response.error_code, response.error_message) {
377        return updates
378            .keys()
379            .map(|feature| (feature.clone(), Err(error.clone())))
380            .collect();
381    }
382    if response.results.is_empty() {
383        return updates
384            .keys()
385            .map(|feature| (feature.clone(), Ok(())))
386            .collect();
387    }
388    let mut results = BTreeMap::new();
389    for result in response.results {
390        if !updates.contains_key(&result.feature) {
391            tracing::warn!(
392                feature = %result.feature,
393                "server response mentioned unknown feature"
394            );
395            continue;
396        }
397        let outcome = kafka_error_if(result.error_code, result.error_message).map_or(Ok(()), Err);
398        results.insert(result.feature, outcome);
399    }
400    for feature in updates.keys() {
401        results.entry(feature.clone()).or_insert_with(|| {
402            Err(KafkaError {
403                code: UNKNOWN_SERVER_ERROR,
404                name: kafka_error_name(UNKNOWN_SERVER_ERROR),
405                message: Some(format!(
406                    "The controller response did not contain a result for feature {feature}"
407                )),
408            })
409        });
410    }
411    results
412}
413
414#[cfg(test)]
415mod tests {
416    use std::{
417        net::SocketAddr,
418        sync::{Arc, Mutex},
419    };
420
421    use krabka_client_core::{ClientError, MockBroker, MockReply};
422    use krabka_protocol::owned::{
423        api_versions_request,
424        api_versions_response::{ApiVersion, FinalizedFeatureKey, SupportedFeatureKey},
425        metadata_request,
426        metadata_response::{MetadataResponse, MetadataResponseBroker},
427        update_features_request,
428        update_features_response::UpdatableFeatureResult,
429    };
430
431    use super::*;
432    use crate::{
433        AdminClientLike as _,
434        partition_leaders::test_support::{
435            ApiVersionRange, api_versions, decode_request, encode_response, fast_admin,
436        },
437        retry::REQUEST_TIMED_OUT,
438    };
439
440    fn update(level: i16, upgrade_type: UpgradeType) -> FeatureUpdate {
441        FeatureUpdate::new(level, upgrade_type).expect("valid update")
442    }
443
444    fn key(feature: &str, max_version_level: i16, upgrade_type: i8) -> FeatureUpdateKey {
445        FeatureUpdateKey {
446            feature: feature.into(),
447            max_version_level,
448            upgrade_type,
449            ..Default::default()
450        }
451    }
452
453    fn error(code: i16, message: Option<&str>) -> KafkaError {
454        KafkaError {
455            code,
456            name: kafka_error_name(code),
457            message: message.map(Into::into),
458        }
459    }
460
461    #[test]
462    fn feature_update_refuses_what_kafka_refuses() {
463        for (name, level, upgrade_type, ok) in [
464            ("upgrade to 1", 1, UpgradeType::Upgrade, true),
465            ("safe delete", 0, UpgradeType::SafeDowngrade, true),
466            ("unsafe delete", 0, UpgradeType::UnsafeDowngrade, true),
467            ("upgrade to 0", 0, UpgradeType::Upgrade, false),
468            ("negative level", -1, UpgradeType::SafeDowngrade, false),
469        ] {
470            let result = FeatureUpdate::new(level, upgrade_type);
471            match (ok, result) {
472                (true, Ok(update)) => assert2::assert!(
473                    (update.max_version_level(), update.upgrade_type()) == (level, upgrade_type),
474                    "case {name}"
475                ),
476                (false, Err(AdminError::InvalidArgument(_))) => {}
477                (_, other) => panic!("case {name}: {other:?}"),
478            }
479        }
480    }
481
482    #[test]
483    fn upgrade_type_codes_match_kafka() {
484        assert2::assert!(
485            [
486                UpgradeType::Upgrade.code(),
487                UpgradeType::SafeDowngrade.code(),
488                UpgradeType::UnsafeDowngrade.code(),
489            ] == [1, 2, 3]
490        );
491    }
492
493    #[test]
494    fn describe_features_preserves_supported_finalized_and_epoch() {
495        for (name, epoch, expected_epoch) in
496            [("finalized", 7, Some(7)), ("no finalized epoch", -1, None)]
497        {
498            let metadata = parse_feature_metadata(ApiVersionsResponse {
499                supported_features: vec![SupportedFeatureKey {
500                    name: "kraft.version".into(),
501                    min_version: 0,
502                    max_version: 1,
503                    ..Default::default()
504                }],
505                finalized_features: vec![FinalizedFeatureKey {
506                    name: "kraft.version".into(),
507                    min_version_level: 1,
508                    max_version_level: 1,
509                    ..Default::default()
510                }],
511                finalized_features_epoch: epoch,
512                ..Default::default()
513            })
514            .unwrap();
515            assert2::assert!(
516                metadata
517                    == FeatureMetadata {
518                        supported: vec![FeatureRange {
519                            name: "kraft.version".into(),
520                            min_version: 0,
521                            max_version: 1,
522                        }],
523                        finalized: vec![FeatureRange {
524                            name: "kraft.version".into(),
525                            min_version: 1,
526                            max_version: 1,
527                        }],
528                        finalized_features_epoch: expected_epoch,
529                    },
530                "case {name}"
531            );
532        }
533    }
534
535    #[test]
536    fn update_results_complete_every_feature_as_kafka_does() {
537        let updates = maplit::btreemap! {
538            "group.version".to_owned() => update(1, UpgradeType::Upgrade),
539            "metadata.version".to_owned() => update(20, UpgradeType::Upgrade),
540        };
541        let row = |feature: &str, error_code, message: Option<&str>| UpdatableFeatureResult {
542            feature: feature.into(),
543            error_code,
544            error_message: message.map(Into::into),
545            ..Default::default()
546        };
547        let missing = "The controller response did not contain a result for feature \
548                       metadata.version";
549        for (name, response, expected) in [
550            (
551                "v2 success has no rows",
552                UpdateFeaturesResponse::default(),
553                [Ok(()), Ok(())],
554            ),
555            (
556                "top-level error fails every feature",
557                UpdateFeaturesResponse {
558                    error_code: 95,
559                    error_message: Some("bad".into()),
560                    ..Default::default()
561                },
562                [Err(error(95, Some("bad"))), Err(error(95, Some("bad")))],
563            ),
564            (
565                "per-feature rows, one missing, one unknown",
566                UpdateFeaturesResponse {
567                    results: vec![
568                        row("group.version", 95, Some("no")),
569                        row("other.version", 0, None),
570                    ],
571                    ..Default::default()
572                },
573                [
574                    Err(error(95, Some("no"))),
575                    Err(error(UNKNOWN_SERVER_ERROR, Some(missing))),
576                ],
577            ),
578        ] {
579            let [group, metadata] = expected;
580            assert2::assert!(
581                update_results(&updates, response)
582                    == maplit::btreemap! {
583                        "group.version".to_owned() => group,
584                        "metadata.version".to_owned() => metadata,
585                    },
586                "case {name}"
587            );
588        }
589    }
590
591    /// A broker that advertises `UpdateFeatures` v0 up to `max_version`,
592    /// records each `UpdateFeatures` request with its version, and answers
593    /// with no error.
594    async fn update_features_broker(
595        max_version: i16,
596        seen: Arc<Mutex<Vec<(i16, UpdateFeaturesRequest)>>>,
597    ) -> MockBroker {
598        MockBroker::start(move |api_key, version, _, body| match api_key {
599            api_versions_request::API_KEY => Some(
600                api_versions()
601                    .apis(vec![
602                        ApiVersionRange::builder()
603                            .api_key(update_features_request::API_KEY)
604                            .max_version(max_version)
605                            .build(),
606                    ])
607                    .call(),
608            ),
609            update_features_request::API_KEY => {
610                let mut request: UpdateFeaturesRequest = decode_request(body, version, true);
611                // The deadline that remains differs between runs.
612                request.timeout_ms = 0;
613                seen.lock().expect("seen lock").push((version, request));
614                Some(encode_response(
615                    &UpdateFeaturesResponse::default(),
616                    version,
617                    true,
618                ))
619            }
620            _ => None,
621        })
622        .await
623        .unwrap()
624    }
625
626    /// The whole `UpdateFeatures` request that reaches the controller, for
627    /// each version the controller advertises, each upgrade type and each
628    /// `validate_only`. A downgrade or a validate-only call is refused
629    /// before sending against a v0-only controller, as Kafka's v0 serializer
630    /// throws `UnsupportedVersionException` for a non-default `UpgradeType`
631    /// or `ValidateOnly`. `AllowDowngrade` is never set, as Kafka never sets
632    /// it.
633    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
634    async fn update_features_sends_kafkas_request_per_version() {
635        let upgrade_types = [
636            UpgradeType::Upgrade,
637            UpgradeType::SafeDowngrade,
638            UpgradeType::UnsafeDowngrade,
639        ];
640        for max_version in 0..=2 {
641            for upgrade_type in upgrade_types {
642                for validate_only in [false, true] {
643                    let name = format!("v{max_version} {upgrade_type:?} validate={validate_only}");
644                    let seen = Arc::new(Mutex::new(Vec::new()));
645                    let broker = update_features_broker(max_version, Arc::clone(&seen)).await;
646                    let mut admin = AdminClient::connect(&[broker.addr.to_string()])
647                        .await
648                        .expect("admin connects");
649                    let level = if upgrade_type == UpgradeType::Upgrade {
650                        7
651                    } else {
652                        0
653                    };
654                    let updates = maplit::btreemap! {
655                        "group.version".to_owned() => update(1, upgrade_type),
656                        "share.version".to_owned() => update(level, upgrade_type),
657                    };
658
659                    let result = admin
660                        .update_features(
661                            &updates,
662                            UpdateFeaturesOptions::builder()
663                                .validate_only(validate_only)
664                                .build(),
665                        )
666                        .await;
667                    broker.stop();
668
669                    let sendable = max_version >= 1
670                        || (upgrade_type == UpgradeType::Upgrade && !validate_only);
671                    if !sendable {
672                        assert2::assert!(
673                            matches!(
674                                result,
675                                Err(AdminError::Transport(ClientError::IncompatibleVersion {
676                                    api_key: update_features_request::API_KEY,
677                                    client_min: 1,
678                                    ..
679                                }))
680                            ),
681                            "case {name}: {result:?}"
682                        );
683                        assert2::assert!(seen.lock().unwrap().is_empty(), "case {name}");
684                        continue;
685                    }
686                    let code = upgrade_type.code();
687                    // v0 carries neither field, so it decodes as the defaults.
688                    let (wire_code, wire_validate) = if max_version == 0 {
689                        (1, false)
690                    } else {
691                        (code, validate_only)
692                    };
693                    assert2::assert!(
694                        result.expect("update succeeds")
695                            == maplit::btreemap! {
696                                "group.version".to_owned() => Ok(()),
697                                "share.version".to_owned() => Ok(()),
698                            },
699                        "case {name}"
700                    );
701                    assert2::assert!(
702                        *seen.lock().unwrap()
703                            == vec![(
704                                max_version,
705                                UpdateFeaturesRequest {
706                                    timeout_ms: 0,
707                                    feature_updates: vec![
708                                        key("group.version", 1, wire_code),
709                                        key("share.version", level, wire_code),
710                                    ],
711                                    validate_only: wire_validate,
712                                    ..Default::default()
713                                }
714                            )],
715                        "case {name}"
716                    );
717                }
718            }
719        }
720    }
721
722    #[tokio::test]
723    async fn update_features_refuses_empty_or_blank_features_before_sending() {
724        let broker = MockBroker::start(|api_key, _, _, _| {
725            (api_key == api_versions_request::API_KEY).then(|| {
726                api_versions()
727                    .apis(vec![
728                        ApiVersionRange::builder()
729                            .api_key(update_features_request::API_KEY)
730                            .max_version(2)
731                            .build(),
732                    ])
733                    .call()
734            })
735        })
736        .await
737        .unwrap();
738        let mut admin = AdminClient::connect(&[broker.addr.to_string()])
739            .await
740            .expect("admin connects");
741        for (name, updates) in [
742            ("empty", BTreeMap::new()),
743            (
744                "blank",
745                maplit::btreemap! {
746                    " ".to_owned() => update(1, UpgradeType::Upgrade),
747                },
748            ),
749        ] {
750            let result = admin
751                .update_features(&updates, UpdateFeaturesOptions::builder().build())
752                .await;
753            assert2::assert!(
754                matches!(result, Err(AdminError::InvalidArgument(_))),
755                "case {name}"
756            );
757        }
758        broker.stop();
759    }
760
761    /// A broker `node_id` that answers `ApiVersions` with `supported` as its
762    /// one supported feature, and `Metadata` with `nodes`.
763    async fn feature_node(
764        supported: &'static str,
765        nodes: Arc<Mutex<Vec<(i32, SocketAddr)>>>,
766    ) -> MockBroker {
767        MockBroker::start_with_replies(move |api_key, version, _, _| match api_key {
768            api_versions_request::API_KEY => MockReply::Respond(encode_response(
769                &ApiVersionsResponse {
770                    api_keys: vec![
771                        ApiVersion {
772                            api_key: api_versions_request::API_KEY,
773                            min_version: 0,
774                            max_version: 4,
775                            ..Default::default()
776                        },
777                        ApiVersion {
778                            api_key: metadata_request::API_KEY,
779                            min_version: 12,
780                            max_version: 12,
781                            ..Default::default()
782                        },
783                    ],
784                    supported_features: vec![SupportedFeatureKey {
785                        name: supported.into(),
786                        min_version: 0,
787                        max_version: 1,
788                        ..Default::default()
789                    }],
790                    finalized_features_epoch: 3,
791                    ..Default::default()
792                },
793                version,
794                false,
795            )),
796            metadata_request::API_KEY => MockReply::Respond(encode_response(
797                &MetadataResponse {
798                    controller_id: 1,
799                    brokers: nodes
800                        .lock()
801                        .expect("nodes lock")
802                        .iter()
803                        .map(|(node_id, addr)| MetadataResponseBroker {
804                            node_id: *node_id,
805                            host: addr.ip().to_string(),
806                            port: i32::from(addr.port()),
807                            ..Default::default()
808                        })
809                        .collect(),
810                    ..Default::default()
811                },
812                version,
813                true,
814            )),
815            _ => MockReply::Silent,
816        })
817        .await
818        .unwrap()
819    }
820
821    /// `describe_features` asks the node of the client's connection, or the
822    /// node that `node_id` names, as Kafka's `describeFeatures` with a
823    /// `DescribeFeaturesOptions.nodeId` does for `kafka-features describe
824    /// --node-id`.
825    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
826    async fn describe_features_asks_the_named_node() {
827        let nodes = Arc::new(Mutex::new(Vec::new()));
828        let first = feature_node("on.node.1", Arc::clone(&nodes)).await;
829        let second = feature_node("on.node.2", Arc::clone(&nodes)).await;
830        *nodes.lock().unwrap() = vec![(1, first.addr), (2, second.addr)];
831        let admin = fast_admin(first.addr, krabka_units::secs(5)).await;
832
833        for (name, node_id, feature) in [
834            ("any node", None, "on.node.1"),
835            ("node 1", Some(1), "on.node.1"),
836            ("node 2", Some(2), "on.node.2"),
837        ] {
838            let metadata = admin
839                .describe_features(
840                    DescribeFeaturesOptions::builder()
841                        .maybe_node_id(node_id)
842                        .build(),
843                )
844                .await
845                .expect("describe succeeds");
846            assert2::assert!(
847                metadata
848                    == FeatureMetadata {
849                        supported: vec![FeatureRange {
850                            name: feature.into(),
851                            min_version: 0,
852                            max_version: 1,
853                        }],
854                        finalized: Vec::new(),
855                        finalized_features_epoch: Some(3),
856                    },
857                "case {name}"
858            );
859        }
860
861        let unknown = admin
862            .describe_features(
863                DescribeFeaturesOptions::builder()
864                    .timeout(50)
865                    .node_id(9)
866                    .build(),
867            )
868            .await;
869        assert2::assert!(
870            matches!(
871                unknown,
872                Err(AdminError::Broker {
873                    code: REQUEST_TIMED_OUT,
874                    ..
875                })
876            ),
877            "{unknown:?}"
878        );
879        first.stop();
880        second.stop();
881    }
882
883    #[tokio::test]
884    async fn update_metadata_version_surfaces_the_feature_error() {
885        let broker = MockBroker::start(|api_key, version, _, _| match api_key {
886            api_versions_request::API_KEY => Some(
887                api_versions()
888                    .apis(vec![
889                        ApiVersionRange::builder()
890                            .api_key(update_features_request::API_KEY)
891                            .max_version(1)
892                            .build(),
893                    ])
894                    .call(),
895            ),
896            update_features_request::API_KEY => Some(encode_response(
897                &UpdateFeaturesResponse {
898                    results: vec![UpdatableFeatureResult {
899                        feature: METADATA_VERSION_FEATURE.into(),
900                        error_code: 95,
901                        error_message: Some("not supported by every broker".into()),
902                        ..Default::default()
903                    }],
904                    ..Default::default()
905                },
906                version,
907                true,
908            )),
909            _ => None,
910        })
911        .await
912        .unwrap();
913        let mut admin = AdminClient::connect(&[broker.addr.to_string()])
914            .await
915            .expect("admin connects");
916        let result = admin
917            .update_metadata_version(15, UpgradeType::SafeDowngrade, krabka_units::secs(5))
918            .await;
919        broker.stop();
920        assert2::assert!(
921            matches!(
922                result,
923                Err(AdminError::Broker {
924                    api: "UpdateFeatures",
925                    code: 95,
926                    ..
927                })
928            ),
929            "{result:?}"
930        );
931    }
932}
krabka_client_admin/features.rs

krabka_client_admin/
features.rs
