krabka_client_admin/
configs.rs

1//! Config reads and changes of any config resource, as Kafka's
2//! `Admin.describeConfigs` and `Admin.incrementalAlterConfigs`.
3//!
4//! Both calls route each resource as `KafkaAdminClient` does (`nodeFor`). A
5//! `BROKER` resource that names a broker id, and every `BROKER_LOGGER`
6//! resource, go to that node in a request of their own
7//! (`ConstantNodeIdProvider`). Every other resource, including the empty-name
8//! `BROKER` resource of the cluster-wide defaults, goes in one request on the
9//! client's connection (`LeastLoadedBrokerOrActiveKController`). With a
10//! controller bootstrap (KIP-919), `incrementalAlterConfigs` sends a `BROKER`
11//! resource with the others to the active controller too.
12//!
13//! Each resource gets a result of its own: its config or change, or the
14//! Kafka error of the resource or of the request that carried it.
15
16use std::{
17    collections::{BTreeMap, BTreeSet},
18    fmt, iter,
19};
20
21use futures_util::future::join_all;
22use krabka_protocol::owned::{
23    describe_configs_request::{DescribeConfigsRequest, DescribeConfigsResource},
24    describe_configs_response::{
25        DescribeConfigsResourceResult, DescribeConfigsResponse, DescribeConfigsResult,
26    },
27    incremental_alter_configs_request::{
28        AlterConfigsResource, AlterableConfig, IncrementalAlterConfigsRequest,
29    },
30    incremental_alter_configs_response::IncrementalAlterConfigsResponse,
31};
32
33use crate::{
34    AdminClient, AdminError, KafkaError, NOT_CONTROLLER,
35    config_resources::{ConfigResource, ConfigResourceType},
36    groups::list_groups_kafka_error,
37    kafka_error_name,
38    log_dirs::{NodeTarget, call_node},
39    retry::{
40        ControllerRetry, CoordinatorRetry, REQUEST_TIMED_OUT, call_timeout_error,
41        is_connection_failure,
42    },
43};
44
45/// `UNKNOWN_SERVER_ERROR`: Kafka's error for an exception that is not an
46/// `ApiException`.
47const UNKNOWN_SERVER_ERROR: i16 = -1;
48/// `NOT_LEADER_OR_FOLLOWER`, which a follower controller gives (KIP-919).
49const NOT_LEADER_OR_FOLLOWER: i16 = 6;
50/// The first `DescribeConfigs` version that carries `IncludeDocumentation`.
51const DESCRIBE_CONFIGS_DOCUMENTATION_VERSION: i16 = 3;
52
53impl ConfigResource {
54    /// The resource of `resource_type` named `name`.
55    #[must_use]
56    pub fn new(resource_type: ConfigResourceType, name: impl Into<String>) -> Self {
57        Self {
58            resource_type,
59            name: name.into(),
60        }
61    }
62
63    /// The topic `name`.
64    #[must_use]
65    pub fn topic(name: impl Into<String>) -> Self {
66        Self::new(ConfigResourceType::Topic, name)
67    }
68
69    /// The broker `broker_id`.
70    #[must_use]
71    pub fn broker(broker_id: i32) -> Self {
72        Self::new(ConfigResourceType::Broker, broker_id.to_string())
73    }
74
75    /// The cluster-wide default broker config: the `BROKER` resource with an
76    /// empty name.
77    #[must_use]
78    pub fn broker_default() -> Self {
79        Self::new(ConfigResourceType::Broker, "")
80    }
81
82    /// The loggers of broker `broker_id`.
83    #[must_use]
84    pub fn broker_logger(broker_id: i32) -> Self {
85        Self::new(ConfigResourceType::BrokerLogger, broker_id.to_string())
86    }
87
88    /// The client-metrics subscription `name` (KIP-714).
89    #[must_use]
90    pub fn client_metrics(name: impl Into<String>) -> Self {
91        Self::new(ConfigResourceType::ClientMetrics, name)
92    }
93
94    /// The group `name` (KIP-848).
95    #[must_use]
96    pub fn group(name: impl Into<String>) -> Self {
97        Self::new(ConfigResourceType::Group, name)
98    }
99
100    /// Whether the resource is a default one, that is has an empty name, as
101    /// Kafka's `ConfigResource.isDefault`.
102    #[must_use]
103    pub fn is_default(&self) -> bool {
104        self.name.is_empty()
105    }
106
107    /// The source of the configs that are set on this resource itself, as
108    /// `kafka-configs --describe` filters them: `DYNAMIC_TOPIC_CONFIG` for a
109    /// topic, `DYNAMIC_BROKER_CONFIG` for a broker,
110    /// `DYNAMIC_DEFAULT_BROKER_CONFIG` for the default broker,
111    /// `DYNAMIC_BROKER_LOGGER_CONFIG` for broker loggers,
112    /// `DYNAMIC_CLIENT_METRICS_CONFIG` and `DYNAMIC_GROUP_CONFIG`. `None` for
113    /// an unknown type.
114    #[must_use]
115    pub fn dynamic_config_source(&self) -> Option<ConfigSource> {
116        match self.resource_type {
117            ConfigResourceType::Topic => Some(ConfigSource::DynamicTopicConfig),
118            ConfigResourceType::Broker if self.is_default() => {
119                Some(ConfigSource::DynamicDefaultBrokerConfig)
120            }
121            ConfigResourceType::Broker => Some(ConfigSource::DynamicBrokerConfig),
122            ConfigResourceType::BrokerLogger => Some(ConfigSource::DynamicBrokerLoggerConfig),
123            ConfigResourceType::ClientMetrics => Some(ConfigSource::DynamicClientMetricsConfig),
124            ConfigResourceType::Group => Some(ConfigSource::DynamicGroupConfig),
125            ConfigResourceType::Unknown => None,
126        }
127    }
128}
129
130/// Where the value of a config comes from, as Kafka's
131/// `ConfigEntry.ConfigSource`.
132#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
133pub enum ConfigSource {
134    /// Set on the topic (wire id 1).
135    DynamicTopicConfig,
136    /// Set on the loggers of the broker (wire id 6).
137    DynamicBrokerLoggerConfig,
138    /// Set on the broker (wire id 2).
139    DynamicBrokerConfig,
140    /// Set as the default of every broker of the cluster (wire id 3).
141    DynamicDefaultBrokerConfig,
142    /// Set on the client-metrics subscription (wire id 7, KIP-714).
143    DynamicClientMetricsConfig,
144    /// Set on the group (wire id 8, KIP-848).
145    DynamicGroupConfig,
146    /// The broker's static config, such as `server.properties` (wire id 4).
147    StaticBrokerConfig,
148    /// The built-in default (wire id 5).
149    DefaultConfig,
150    /// A source that Kafka does not define (wire id 0, or an id above 8).
151    Unknown,
152}
153
154impl ConfigSource {
155    /// The wire id of the source, as `DescribeConfigsResponse.ConfigSource`.
156    #[must_use]
157    pub const fn id(self) -> i8 {
158        match self {
159            Self::Unknown => 0,
160            Self::DynamicTopicConfig => 1,
161            Self::DynamicBrokerConfig => 2,
162            Self::DynamicDefaultBrokerConfig => 3,
163            Self::StaticBrokerConfig => 4,
164            Self::DefaultConfig => 5,
165            Self::DynamicBrokerLoggerConfig => 6,
166            Self::DynamicClientMetricsConfig => 7,
167            Self::DynamicGroupConfig => 8,
168        }
169    }
170
171    /// The source of a wire id, as `DescribeConfigsResponse.ConfigSource.forId`:
172    /// [`Self::Unknown`] for an id above 8, and `None` for a negative id,
173    /// which Kafka refuses.
174    #[must_use]
175    pub const fn from_id(id: i8) -> Option<Self> {
176        Some(match id {
177            i8::MIN..=-1 => return None,
178            1 => Self::DynamicTopicConfig,
179            2 => Self::DynamicBrokerConfig,
180            3 => Self::DynamicDefaultBrokerConfig,
181            4 => Self::StaticBrokerConfig,
182            5 => Self::DefaultConfig,
183            6 => Self::DynamicBrokerLoggerConfig,
184            7 => Self::DynamicClientMetricsConfig,
185            8 => Self::DynamicGroupConfig,
186            _ => Self::Unknown,
187        })
188    }
189}
190
191/// The data type of a config, as Kafka's `ConfigEntry.ConfigType`.
192#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
193pub enum ConfigType {
194    /// Not reported (`DescribeConfigs` below v3) or not defined (wire id 0,
195    /// or an id above 9).
196    Unknown,
197    /// Wire id 1.
198    Boolean,
199    /// Wire id 2.
200    String,
201    /// Wire id 3.
202    Int,
203    /// Wire id 4.
204    Short,
205    /// Wire id 5.
206    Long,
207    /// Wire id 6.
208    Double,
209    /// Wire id 7.
210    List,
211    /// Wire id 8.
212    Class,
213    /// Wire id 9.
214    Password,
215}
216
217impl ConfigType {
218    /// The wire id of the type, as `DescribeConfigsResponse.ConfigType`.
219    #[must_use]
220    pub const fn id(self) -> i8 {
221        match self {
222            Self::Unknown => 0,
223            Self::Boolean => 1,
224            Self::String => 2,
225            Self::Int => 3,
226            Self::Short => 4,
227            Self::Long => 5,
228            Self::Double => 6,
229            Self::List => 7,
230            Self::Class => 8,
231            Self::Password => 9,
232        }
233    }
234
235    /// The type of a wire id, as `DescribeConfigsResponse.ConfigType.forId`:
236    /// [`Self::Unknown`] for an id above 9, and `None` for a negative id,
237    /// which Kafka refuses.
238    #[must_use]
239    pub const fn from_id(id: i8) -> Option<Self> {
240        Some(match id {
241            i8::MIN..=-1 => return None,
242            1 => Self::Boolean,
243            2 => Self::String,
244            3 => Self::Int,
245            4 => Self::Short,
246            5 => Self::Long,
247            6 => Self::Double,
248            7 => Self::List,
249            8 => Self::Class,
250            9 => Self::Password,
251            _ => Self::Unknown,
252        })
253    }
254}
255
256/// One value that a config takes from one source, as Kafka's
257/// `ConfigEntry.ConfigSynonym`.
258#[derive(Debug, Clone, PartialEq, Eq)]
259pub struct ConfigSynonym {
260    /// The name of the config at that source, such as `log.retention.ms`
261    /// for the topic config `retention.ms`.
262    pub name: String,
263    /// The value at that source. `None` when the broker withholds it, as it
264    /// does for a sensitive config.
265    pub value: Option<String>,
266    /// The source.
267    pub source: ConfigSource,
268}
269
270/// One config of a resource, as Kafka's `ConfigEntry`.
271///
272/// `Debug` prints a sensitive value as `Redacted`, as Kafka's `toString`
273/// does.
274#[derive(Clone, PartialEq, Eq)]
275pub struct ConfigEntry {
276    /// The name of the config.
277    pub name: String,
278    /// The value, as the broker gave it. The broker gives `None` for a
279    /// sensitive config and for a config with no value.
280    pub value: Option<String>,
281    /// Where the value comes from.
282    pub source: ConfigSource,
283    /// Whether the config is sensitive, such as a password.
284    pub is_sensitive: bool,
285    /// Whether the config cannot be changed.
286    pub is_read_only: bool,
287    /// The value of the config at each source, most specific first. Empty
288    /// unless [`DescribeConfigsOptions::include_synonyms`] is set.
289    pub synonyms: Vec<ConfigSynonym>,
290    /// The data type. [`ConfigType::Unknown`] below `DescribeConfigs` v3.
291    pub config_type: ConfigType,
292    /// The documentation. `None` unless
293    /// [`DescribeConfigsOptions::include_documentation`] is set.
294    pub documentation: Option<String>,
295}
296
297impl ConfigEntry {
298    /// Whether the value is the built-in default, as Kafka's
299    /// `ConfigEntry.isDefault`.
300    #[must_use]
301    pub fn is_default(&self) -> bool {
302        self.source == ConfigSource::DefaultConfig
303    }
304}
305
306impl fmt::Debug for ConfigEntry {
307    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
308        let value: &dyn fmt::Debug = if self.is_sensitive {
309            &"Redacted"
310        } else {
311            &self.value
312        };
313        f.debug_struct("ConfigEntry")
314            .field("name", &self.name)
315            .field("value", value)
316            .field("source", &self.source)
317            .field("is_sensitive", &self.is_sensitive)
318            .field("is_read_only", &self.is_read_only)
319            .field("synonyms", &self.synonyms)
320            .field("config_type", &self.config_type)
321            .field("documentation", &self.documentation)
322            .finish()
323    }
324}
325
326/// The configs of one resource, as Kafka's `Config`.
327#[derive(Debug, Clone, Default, PartialEq, Eq)]
328pub struct Config {
329    /// Each config by name. A name the broker lists twice keeps its last
330    /// entry, as Kafka's map does.
331    pub entries: BTreeMap<String, ConfigEntry>,
332}
333
334impl Config {
335    /// The config `name`.
336    #[must_use]
337    pub fn get(&self, name: &str) -> Option<&ConfigEntry> {
338        self.entries.get(name)
339    }
340
341    /// The name and value of each config that is set on `resource` itself:
342    /// the entries whose source is [`ConfigResource::dynamic_config_source`]
343    /// and that have a value. This is what `kafka-configs --describe` lists
344    /// without `--all`, less the sensitive configs, whose value the broker
345    /// withholds.
346    #[must_use]
347    pub fn dynamic_overrides(&self, resource: &ConfigResource) -> BTreeMap<String, String> {
348        let Some(source) = resource.dynamic_config_source() else {
349            return BTreeMap::new();
350        };
351        self.entries
352            .values()
353            .filter(|entry| entry.source == source)
354            .filter_map(|entry| Some((entry.name.clone(), entry.value.clone()?)))
355            .collect()
356    }
357}
358
359/// What [`AdminClientLike::describe_configs`](crate::AdminClientLike::describe_configs) asks for, as Kafka's
360/// `DescribeConfigsOptions`.
361#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, bon::Builder)]
362pub struct DescribeConfigsOptions {
363    /// Ask for the value of each config at each source
364    /// ([`ConfigEntry::synonyms`]).
365    #[builder(default)]
366    pub include_synonyms: bool,
367    /// Ask for the documentation of each config. Needs `DescribeConfigs` v3.
368    #[builder(default)]
369    pub include_documentation: bool,
370}
371
372/// What an [`AlterConfigOp`] does, as Kafka's `AlterConfigOp.OpType`.
373#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
374pub enum AlterConfigOpType {
375    /// Set the value (wire id 0).
376    Set,
377    /// Revert to the default value (wire id 1).
378    Delete,
379    /// Add the values to a list config (wire id 2).
380    Append,
381    /// Remove the values from a list config (wire id 3).
382    Subtract,
383}
384
385impl AlterConfigOpType {
386    /// The wire id of the operation.
387    #[must_use]
388    pub const fn id(self) -> i8 {
389        match self {
390            Self::Set => 0,
391            Self::Delete => 1,
392            Self::Append => 2,
393            Self::Subtract => 3,
394        }
395    }
396}
397
398/// One change of one config, as Kafka's `AlterConfigOp`.
399#[derive(Debug, Clone, PartialEq, Eq)]
400pub struct AlterConfigOp {
401    /// The name of the config.
402    pub name: String,
403    /// The value that the operation sends. A `Delete` needs none.
404    pub value: Option<String>,
405    /// The operation.
406    pub op_type: AlterConfigOpType,
407}
408
409impl AlterConfigOp {
410    /// Set `name` to `value`.
411    #[must_use]
412    pub fn set(name: impl Into<String>, value: impl Into<String>) -> Self {
413        Self {
414            name: name.into(),
415            value: Some(value.into()),
416            op_type: AlterConfigOpType::Set,
417        }
418    }
419
420    /// Revert `name` to its default value. The operation sends no value.
421    #[must_use]
422    pub fn delete(name: impl Into<String>) -> Self {
423        Self {
424            name: name.into(),
425            value: None,
426            op_type: AlterConfigOpType::Delete,
427        }
428    }
429
430    /// Add the comma-separated `values` to the list config `name`.
431    #[must_use]
432    pub fn append(name: impl Into<String>, values: impl Into<String>) -> Self {
433        Self {
434            name: name.into(),
435            value: Some(values.into()),
436            op_type: AlterConfigOpType::Append,
437        }
438    }
439
440    /// Remove the comma-separated `values` from the list config `name`.
441    #[must_use]
442    pub fn subtract(name: impl Into<String>, values: impl Into<String>) -> Self {
443        Self {
444            name: name.into(),
445            value: Some(values.into()),
446            op_type: AlterConfigOpType::Subtract,
447        }
448    }
449}
450
451/// What [`AdminClientLike::incremental_alter_configs`](crate::AdminClientLike::incremental_alter_configs) asks for, as Kafka's
452/// `AlterConfigsOptions`.
453#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, bon::Builder)]
454pub struct IncrementalAlterConfigsOptions {
455    /// Validate the changes and apply none.
456    #[builder(default)]
457    pub validate_only: bool,
458}
459
460/// The result of each resource of a [`AdminClientLike::describe_configs`](crate::AdminClientLike::describe_configs) call.
461pub type DescribeConfigsResults = BTreeMap<ConfigResource, Result<Config, KafkaError>>;
462
463/// The result of each resource of an
464/// [`AdminClientLike::incremental_alter_configs`](crate::AdminClientLike::incremental_alter_configs) call.
465pub type AlterConfigsResults = BTreeMap<ConfigResource, Result<(), KafkaError>>;
466
467/// The node that a resource goes to, as `KafkaAdminClient.nodeFor`: the
468/// broker id of a `BROKER` resource with a name and of a `BROKER_LOGGER`
469/// resource, and `None` for every other resource.
470fn node_for(resource: &ConfigResource) -> Result<Option<i32>, AdminError> {
471    let by_id = match resource.resource_type {
472        ConfigResourceType::Broker => !resource.is_default(),
473        ConfigResourceType::BrokerLogger => true,
474        _ => false,
475    };
476    if !by_id {
477        return Ok(None);
478    }
479    resource.name.parse().map(Some).map_err(|_| {
480        AdminError::InvalidArgument(format!(
481            "the name of a {:?} config resource must be a broker id, found {:?}",
482            resource.resource_type, resource.name
483        ))
484    })
485}
486
487/// The Kafka error of a request that failed as a whole. A connection that
488/// fails until the deadline gives `REQUEST_TIMED_OUT` (7), as Kafka's
489/// `Call.handleTimeoutFailure` does.
490fn call_error(api: &str, error: &AdminError) -> KafkaError {
491    if is_connection_failure(error) {
492        return KafkaError {
493            code: REQUEST_TIMED_OUT,
494            name: kafka_error_name(REQUEST_TIMED_OUT),
495            message: Some(format!("{api} timed out: {error}")),
496        };
497    }
498    list_groups_kafka_error(error)
499}
500
501/// `UNKNOWN_SERVER_ERROR` with `message`: Kafka's error for a failure that is
502/// not an `ApiException`.
503fn unknown_server_error(message: String) -> KafkaError {
504    KafkaError {
505        code: UNKNOWN_SERVER_ERROR,
506        name: kafka_error_name(UNKNOWN_SERVER_ERROR),
507        message: Some(message),
508    }
509}
510
511fn describe_request(
512    resources: &BTreeSet<ConfigResource>,
513    options: DescribeConfigsOptions,
514) -> DescribeConfigsRequest {
515    DescribeConfigsRequest {
516        resources: resources
517            .iter()
518            .map(|resource| DescribeConfigsResource {
519                resource_type: resource.resource_type.id(),
520                resource_name: resource.name.clone(),
521                configuration_keys: None,
522                ..Default::default()
523            })
524            .collect(),
525        include_synonyms: options.include_synonyms,
526        include_documentation: options.include_documentation,
527        ..Default::default()
528    }
529}
530
531/// The lowest `DescribeConfigs` version that carries `options`.
532/// `IncludeDocumentation` is not ignorable, so Kafka refuses to send it below
533/// v3.
534fn describe_min_version(options: DescribeConfigsOptions) -> Option<i16> {
535    options
536        .include_documentation
537        .then_some(DESCRIBE_CONFIGS_DOCUMENTATION_VERSION)
538}
539
540/// The source of a wire id, or Kafka's `IllegalArgumentException` message.
541fn source_of(id: i8) -> Result<ConfigSource, String> {
542    ConfigSource::from_id(id).ok_or_else(|| format!("id should be positive, id: {id}"))
543}
544
545/// The config of one resource of a `DescribeConfigs` answer, as Kafka's
546/// `describeConfigResult`. A negative source or type id gives Kafka's
547/// `IllegalArgumentException` message.
548fn config_of(configs: Vec<DescribeConfigsResourceResult>) -> Result<Config, String> {
549    let mut entries = BTreeMap::new();
550    for config in configs {
551        let synonyms = config
552            .synonyms
553            .into_iter()
554            .map(|synonym| {
555                Ok(ConfigSynonym {
556                    name: synonym.name,
557                    value: synonym.value,
558                    source: source_of(synonym.source)?,
559                })
560            })
561            .collect::<Result<_, String>>()?;
562        let entry = ConfigEntry {
563            source: source_of(config.config_source)?,
564            config_type: ConfigType::from_id(config.config_type)
565                .ok_or_else(|| format!("id should be positive, id: {}", config.config_type))?,
566            name: config.name,
567            value: config.value,
568            is_sensitive: config.is_sensitive,
569            is_read_only: config.read_only,
570            synonyms,
571            documentation: config.documentation,
572        };
573        entries.insert(entry.name.clone(), entry);
574    }
575    Ok(Config { entries })
576}
577
578/// The result of each resource of `asked` from the answer to one
579/// `DescribeConfigs` request, as Kafka's `describeConfigs` call handles it:
580///
581/// - A resource with an error code gets that error.
582/// - A resource that the answer does not name gets `UNKNOWN_SERVER_ERROR`
583///   (`completeUnrealizedFutures`). A resource that the request did not name
584///   is skipped.
585/// - A request that failed, an answer that names a resource twice, and an
586///   answer with a negative source or type id fail every resource of the
587///   request, as Kafka's `handleFailure` does.
588fn describe_results(
589    asked: &BTreeSet<ConfigResource>,
590    answer: Result<DescribeConfigsResponse, KafkaError>,
591) -> DescribeConfigsResults {
592    let fail_all = |error: KafkaError| {
593        asked
594            .iter()
595            .map(|resource| (resource.clone(), Err(error.clone())))
596            .collect()
597    };
598    let response = match answer {
599        Ok(response) => response,
600        Err(error) => return fail_all(error),
601    };
602    let mut answered = BTreeMap::<ConfigResource, DescribeConfigsResult>::new();
603    for result in response.results {
604        let resource = ConfigResource::new(
605            ConfigResourceType::from_id(result.resource_type),
606            result.resource_name.clone(),
607        );
608        if answered.insert(resource.clone(), result).is_some() {
609            return fail_all(unknown_server_error(format!(
610                "Duplicate key for config resource {resource:?}"
611            )));
612        }
613    }
614    let mut out = BTreeMap::new();
615    for resource in asked {
616        let result = match answered.remove(resource) {
617            None => Err(unknown_server_error(format!(
618                "The node response did not contain a result for config resource {resource:?}"
619            ))),
620            Some(result) if result.error_code != 0 => Err(KafkaError {
621                code: result.error_code,
622                name: kafka_error_name(result.error_code),
623                message: result.error_message,
624            }),
625            Some(result) => match config_of(result.configs) {
626                Ok(config) => Ok(config),
627                Err(message) => return fail_all(unknown_server_error(message)),
628            },
629        };
630        out.insert(resource.clone(), result);
631    }
632    for resource in answered.into_keys() {
633        tracing::warn!(
634            ?resource,
635            "the DescribeConfigs response names a resource that was not asked for"
636        );
637    }
638    out
639}
640
641fn alter_request(
642    resources: &BTreeMap<ConfigResource, &[AlterConfigOp]>,
643    options: IncrementalAlterConfigsOptions,
644) -> IncrementalAlterConfigsRequest {
645    IncrementalAlterConfigsRequest {
646        resources: resources
647            .iter()
648            .map(|(resource, ops)| AlterConfigsResource {
649                resource_type: resource.resource_type.id(),
650                resource_name: resource.name.clone(),
651                configs: ops
652                    .iter()
653                    .map(|op| AlterableConfig {
654                        name: op.name.clone(),
655                        config_operation: op.op_type.id(),
656                        value: op.value.clone(),
657                        ..Default::default()
658                    })
659                    .collect(),
660                ..Default::default()
661            })
662            .collect(),
663        validate_only: options.validate_only,
664        ..Default::default()
665    }
666}
667
668/// Whether an `IncrementalAlterConfigs` answer makes Kafka find the
669/// controller again and retry (`handleNotControllerError`): any
670/// `NOT_CONTROLLER` (41), or with a controller bootstrap any
671/// `NOT_LEADER_OR_FOLLOWER` (6).
672fn names_not_controller(response: &IncrementalAlterConfigsResponse, controllers: bool) -> bool {
673    response.responses.iter().any(|result| {
674        result.error_code == NOT_CONTROLLER
675            || (controllers && result.error_code == NOT_LEADER_OR_FOLLOWER)
676    })
677}
678
679/// The result of each resource of `asked` from the answer to one
680/// `IncrementalAlterConfigs` request. A resource that the answer does not
681/// name gets `UNKNOWN_SERVER_ERROR`. A request that failed fails every
682/// resource.
683fn alter_results<'a>(
684    asked: impl Iterator<Item = &'a ConfigResource>,
685    answer: Result<IncrementalAlterConfigsResponse, KafkaError>,
686) -> AlterConfigsResults {
687    let response = match answer {
688        Ok(response) => response,
689        Err(error) => {
690            return asked
691                .map(|resource| (resource.clone(), Err(error.clone())))
692                .collect();
693        }
694    };
695    let answered = response
696        .responses
697        .into_iter()
698        .map(|result| {
699            (
700                ConfigResource::new(
701                    ConfigResourceType::from_id(result.resource_type),
702                    result.resource_name,
703                ),
704                (result.error_code, result.error_message),
705            )
706        })
707        .collect::<BTreeMap<_, _>>();
708    asked
709        .map(|resource| {
710            let result = match answered.get(resource) {
711                None => Err(unknown_server_error(format!(
712                    "The node response did not contain a result for config resource {resource:?}"
713                ))),
714                Some((0, _)) => Ok(()),
715                Some((code, message)) => Err(KafkaError {
716                    code: *code,
717                    name: kafka_error_name(*code),
718                    message: message.clone(),
719                }),
720            };
721            (resource.clone(), result)
722        })
723        .collect()
724}
725
726impl AdminClient {
727    /// The call behind [`AdminClientLike::describe_configs`](crate::AdminClientLike::describe_configs).
728    pub(crate) async fn describe_configs_impl(
729        &self,
730        resources: &[ConfigResource],
731        options: DescribeConfigsOptions,
732    ) -> Result<DescribeConfigsResults, AdminError> {
733        let mut by_node = BTreeMap::<Option<i32>, BTreeSet<ConfigResource>>::new();
734        for resource in resources {
735            by_node
736                .entry(node_for(resource)?)
737                .or_default()
738                .insert(resource.clone());
739        }
740        let start = self.retry.start();
741        let min_version = describe_min_version(options);
742        let answers = join_all(by_node.iter().map(|(node, asked)| {
743            let request = describe_request(asked, options);
744            async move {
745                let answer = if let Some(node_id) = *node {
746                    call_node(
747                        &self.conn,
748                        NodeTarget {
749                            node_id,
750                            supports_controllers: true,
751                            min_version,
752                        },
753                        &self.options,
754                        request,
755                        CoordinatorRetry::from_deadline(start),
756                    )
757                    .await
758                } else {
759                    let retry = ControllerRetry::new("DescribeConfigs", self.retry);
760                    let sent = match min_version {
761                        Some(min_version) => {
762                            retry
763                                .bounded(self.conn.send_at_least(request, min_version))
764                                .await
765                        }
766                        None => retry.bounded(self.conn.send(request)).await,
767                    };
768                    sent.map_err(|error| call_error("DescribeConfigs", &error))
769                };
770                describe_results(asked, answer)
771            }
772        }))
773        .await;
774        Ok(answers.into_iter().flatten().collect())
775    }
776
777    /// The call behind [`AdminClientLike::incremental_alter_configs`](crate::AdminClientLike::incremental_alter_configs).
778    pub(crate) async fn incremental_alter_configs_impl(
779        &self,
780        configs: &BTreeMap<ConfigResource, Vec<AlterConfigOp>>,
781        options: IncrementalAlterConfigsOptions,
782    ) -> Result<AlterConfigsResults, AdminError> {
783        let controllers = self.conn.uses_controller_bootstrap();
784        let mut unified = BTreeMap::new();
785        let mut by_node = Vec::new();
786        for (resource, ops) in configs {
787            let node = node_for(resource)?.filter(|_| {
788                !controllers || resource.resource_type == ConfigResourceType::BrokerLogger
789            });
790            match node {
791                Some(node_id) => by_node.push((node_id, resource, ops.as_slice())),
792                None => {
793                    unified.insert(resource.clone(), ops.as_slice());
794                }
795            }
796        }
797        let start = self.retry.start();
798        let (conn, client_options) = (&self.conn, &self.options);
799        let node_answers =
800            join_all(by_node.iter().map(|(node_id, resource, ops)| {
801                let request = alter_request(&maplit::btreemap! {
802                    (*resource).clone() => *ops,
803                }, options);
804                let target = NodeTarget {
805                    node_id: *node_id,
806                    supports_controllers: true,
807                    min_version: None,
808                };
809                async move {
810                    let mut deadline = start;
811                    let mut attempts = 1;
812                    let answer = loop {
813                        let answer = call_node(
814                            conn,
815                            target,
816                            client_options,
817                            request.clone(),
818                            CoordinatorRetry::from_deadline(deadline),
819                        )
820                        .await;
821                        if !matches!(&answer, Ok(response) if names_not_controller(response, controllers))
822                        {
823                            break answer;
824                        }
825                        let timed_out = || {
826                            Err(call_timeout_error(
827                                "IncrementalAlterConfigs",
828                                attempts,
829                                "NOT_CONTROLLER",
830                            ))
831                        };
832                        if deadline.exhausted() {
833                            break timed_out();
834                        }
835                        deadline.backoff().await;
836                        if deadline.expired() {
837                            break timed_out();
838                        }
839                        attempts += 1;
840                    };
841                    alter_results(iter::once(*resource), answer)
842                }
843            }))
844            .await;
845        let mut out: AlterConfigsResults = node_answers.into_iter().flatten().collect();
846        if !unified.is_empty() {
847            let answer = self.alter_unified(&unified, options, controllers).await;
848            out.extend(alter_results(unified.keys(), answer));
849        }
850        Ok(out)
851    }
852
853    /// The `IncrementalAlterConfigs` request of the resources that go on the
854    /// client's connection, sent again after each `NOT_CONTROLLER` answer
855    /// until the deadline.
856    async fn alter_unified(
857        &self,
858        resources: &BTreeMap<ConfigResource, &[AlterConfigOp]>,
859        options: IncrementalAlterConfigsOptions,
860        controllers: bool,
861    ) -> Result<IncrementalAlterConfigsResponse, KafkaError> {
862        const API: &str = "IncrementalAlterConfigs";
863        let request = alter_request(resources, options);
864        let mut retry = ControllerRetry::new(API, self.retry);
865        loop {
866            let response = retry
867                .bounded(self.conn.send(request.clone()))
868                .await
869                .map_err(|error| call_error(API, &error))?;
870            if !names_not_controller(&response, controllers) {
871                return Ok(response);
872            }
873            retry
874                .after_not_controller(self)
875                .await
876                .map_err(|error| call_error(API, &error))?;
877        }
878    }
879}
880
881#[cfg(test)]
882mod tests;
krabka_client_admin/configs.rs

krabka_client_admin/
configs.rs
