krabka_client_admin/
config.rs

1//! Admin client settings, as Kafka's `AdminClientConfig` defines them.
2
3use std::sync::atomic::{AtomicU32, Ordering};
4
5use krabka_client_core::{
6    ClientDnsLookup, ClientDnsTimeout, ClientFrameMax, ConnectionDispatchQueueCapacity,
7    ConnectionOptions, DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER, MetadataRecoveryStrategy,
8    security::ClientSecurity,
9};
10use krabka_units::{
11    ByteSize, Time,
12    convert::{ByteSizeExt as _, TimeExt as _},
13    millis, secs,
14};
15
16use crate::{AdminError, retry::RetryPolicy};
17
18/// Kafka's `request.timeout.ms` default for the admin client.
19pub const DEFAULT_ADMIN_REQUEST_TIMEOUT: Time = secs(30);
20/// Kafka's `default.api.timeout.ms` default.
21pub const DEFAULT_API_TIMEOUT: Time = secs(60);
22/// Kafka's `retry.backoff.ms` default.
23pub const DEFAULT_RETRY_BACKOFF: Time = millis(100);
24/// Kafka's `retry.backoff.max.ms` default.
25pub const DEFAULT_RETRY_BACKOFF_MAX: Time = secs(1);
26/// Kafka's `socket.connection.setup.timeout.ms` default.
27pub const DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT: Time = secs(10);
28/// Kafka's `socket.connection.setup.timeout.max.ms` default.
29pub const DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX: Time = secs(30);
30/// Kafka's admin client `connections.max.idle.ms` default (5 minutes).
31pub const DEFAULT_ADMIN_CONNECTIONS_MAX_IDLE: Time = krabka_units::minutes(5);
32/// Kafka's `enable.metrics.push` default.
33pub const DEFAULT_ENABLE_METRICS_PUSH: bool = true;
34
35/// `CommonClientConfigs.RETRY_BACKOFF_JITTER`.
36const RETRY_BACKOFF_JITTER: f64 = 0.2;
37
38/// The sequence of generated client ids, as Kafka's
39/// `KafkaAdminClient.ADMIN_CLIENT_ID_SEQUENCE`. The first id is
40/// `adminclient-1`.
41static ADMIN_CLIENT_ID_SEQUENCE: AtomicU32 = AtomicU32::new(1);
42
43/// The settings of an [`AdminClient`](crate::AdminClient).
44///
45/// `Default` gives Kafka's `AdminClientConfig` defaults. A setting that is
46/// `None` is not set, which matters where Kafka checks whether the user set
47/// a value (`client.id`, `default.api.timeout.ms`).
48#[derive(Clone, Debug)]
49pub struct AdminClientConfig {
50    /// `client.id`. `None` or an empty id gives a new `adminclient-<n>` for
51    /// each client, as `KafkaAdminClient.generateClientId` does.
52    pub client_id: Option<String>,
53    /// `request.timeout.ms`: the deadline of one request.
54    pub request_timeout: Time,
55    /// `default.api.timeout.ms`: the deadline of a call and its retries when
56    /// the call sets no timeout. `None` uses 60 s, raised to
57    /// [`Self::request_timeout`] when that is larger.
58    pub default_api_timeout: Option<Time>,
59    /// `retries`: the number of retries of a call before its deadline.
60    /// Kafka's default is `Integer.MAX_VALUE`.
61    pub retries: u32,
62    /// `retry.backoff.ms`.
63    pub retry_backoff: Time,
64    /// `retry.backoff.max.ms`.
65    pub retry_backoff_max: Time,
66    /// `socket.connection.setup.timeout.ms`: the connection setup deadline
67    /// of the first attempt.
68    pub socket_connection_setup_timeout: Time,
69    /// `socket.connection.setup.timeout.max.ms`: the setup deadline grows
70    /// per failed attempt up to this value.
71    pub socket_connection_setup_timeout_max: Time,
72    /// `connections.max.idle.ms`: a connection with no traffic closes after
73    /// this time.
74    pub connections_max_idle: Time,
75    /// The DNS lookup deadline of one broker address.
76    pub dns_timeout: ClientDnsTimeout,
77    /// Bootstrap DNS address selection policy.
78    pub client_dns_lookup: ClientDnsLookup,
79    /// Initial reconnect delay.
80    pub reconnect_backoff: Time,
81    /// Maximum reconnect delay.
82    pub reconnect_backoff_max: Time,
83    /// Outbound socket send buffer; None selects the OS default.
84    pub send_buffer: Option<ByteSize>,
85    /// Outbound socket receive buffer; None selects the OS default.
86    pub receive_buffer: Option<ByteSize>,
87    /// Capacity of every connection's dispatch queue.
88    pub dispatch_queue_capacity: ConnectionDispatchQueueCapacity,
89    /// Maximum accepted response frame size.
90    pub frame_max: ClientFrameMax,
91    /// `security.protocol` and its TLS and SASL settings. `None` is
92    /// plaintext.
93    pub security: Option<ClientSecurity>,
94    /// `metadata.recovery.strategy` (KIP-1102).
95    pub metadata_recovery_strategy: MetadataRecoveryStrategy,
96    /// `metadata.recovery.rebootstrap.trigger.ms` (KIP-1102).
97    pub metadata_recovery_rebootstrap_trigger: Time,
98    /// `enable.metrics.push` (KIP-714, default `true`): push the client
99    /// metrics that a broker's client metrics subscription names, with a
100    /// terminating push on close. When it is `false`, the client sends
101    /// neither `GetTelemetrySubscriptions` nor `PushTelemetry`, and
102    /// `client_instance_id` fails.
103    pub enable_metrics_push: bool,
104}
105
106impl Default for AdminClientConfig {
107    fn default() -> Self {
108        Self {
109            client_id: None,
110            request_timeout: DEFAULT_ADMIN_REQUEST_TIMEOUT,
111            default_api_timeout: None,
112            retries: u32::MAX,
113            retry_backoff: DEFAULT_RETRY_BACKOFF,
114            retry_backoff_max: DEFAULT_RETRY_BACKOFF_MAX,
115            socket_connection_setup_timeout: DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT,
116            socket_connection_setup_timeout_max: DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX,
117            connections_max_idle: DEFAULT_ADMIN_CONNECTIONS_MAX_IDLE,
118            dns_timeout: ClientDnsTimeout::default(),
119            client_dns_lookup: ClientDnsLookup::default(),
120            reconnect_backoff: krabka_client_core::DEFAULT_RECONNECT_BACKOFF,
121            reconnect_backoff_max: krabka_client_core::DEFAULT_RECONNECT_BACKOFF_MAX,
122            send_buffer: Some(krabka_client_core::DEFAULT_SEND_BUFFER),
123            receive_buffer: Some(krabka_client_core::DEFAULT_RECEIVE_BUFFER),
124            dispatch_queue_capacity: ConnectionDispatchQueueCapacity::default(),
125            frame_max: ClientFrameMax::default(),
126            security: None,
127            metadata_recovery_strategy: MetadataRecoveryStrategy::default(),
128            metadata_recovery_rebootstrap_trigger: DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER,
129            enable_metrics_push: DEFAULT_ENABLE_METRICS_PUSH,
130        }
131    }
132}
133
134/// The connection options and retry policy of one admin client.
135#[derive(Clone, Debug)]
136pub struct ResolvedAdminConfig {
137    pub(crate) connection: ConnectionOptions,
138    pub(crate) retry: RetryPolicy,
139    pub(crate) metadata_recovery_strategy: MetadataRecoveryStrategy,
140    pub(crate) metadata_recovery_rebootstrap_trigger: Time,
141    pub(crate) enable_metrics_push: bool,
142}
143
144impl AdminClientConfig {
145    fn validate_transport(&self) -> Result<(), AdminError> {
146        for (name, value) in [
147            ("reconnect_backoff", self.reconnect_backoff),
148            ("reconnect_backoff_max", self.reconnect_backoff_max),
149        ] {
150            if !value.secs_f64().is_finite() || value < Time::ZERO {
151                return Err(AdminError::InvalidConfig(format!(
152                    "{name} must be finite and nonnegative"
153                )));
154            }
155        }
156        if self.reconnect_backoff_max < self.reconnect_backoff {
157            return Err(AdminError::InvalidConfig(
158                "reconnect_backoff_max is below reconnect_backoff".into(),
159            ));
160        }
161        for (name, value) in [
162            ("send_buffer", self.send_buffer),
163            ("receive_buffer", self.receive_buffer),
164        ] {
165            if let Some(value) = value {
166                let bytes = value.bytes_f64();
167                if !bytes.is_finite()
168                    || bytes < 1.0
169                    || bytes.fract() != 0.0
170                    || bytes > f64::from(i32::MAX)
171                {
172                    return Err(AdminError::InvalidConfig(format!(
173                        "{name} must contain positive whole int32 bytes"
174                    )));
175                }
176            }
177        }
178        Ok(())
179    }
180
181    /// Check the settings and derive the connection options and the call
182    /// retry policy.
183    ///
184    /// # Errors
185    /// Returns [`AdminError::InvalidConfig`] when a timeout or backoff is not
186    /// positive, or when `default_api_timeout` is set below
187    /// `request_timeout`, as Kafka's `configureDefaultApiTimeoutMs` throws a
188    /// `ConfigException`.
189    pub(crate) fn resolve(self) -> Result<ResolvedAdminConfig, AdminError> {
190        for (name, value) in [
191            ("request_timeout", self.request_timeout),
192            (
193                "socket_connection_setup_timeout",
194                self.socket_connection_setup_timeout,
195            ),
196            (
197                "socket_connection_setup_timeout_max",
198                self.socket_connection_setup_timeout_max,
199            ),
200        ] {
201            if !value.secs_f64().is_finite() || value <= Time::ZERO {
202                return Err(AdminError::InvalidConfig(format!(
203                    "{name} must be positive and finite"
204                )));
205            }
206        }
207        if let Some(value) = self.default_api_timeout
208            && (!value.secs_f64().is_finite() || value <= Time::ZERO)
209        {
210            return Err(AdminError::InvalidConfig(
211                "default_api_timeout must be positive and finite".to_owned(),
212            ));
213        }
214        for (name, value) in [
215            ("connections_max_idle", self.connections_max_idle),
216            ("retry_backoff", self.retry_backoff),
217            ("retry_backoff_max", self.retry_backoff_max),
218        ] {
219            if !value.secs_f64().is_finite() || value < Time::ZERO {
220                return Err(AdminError::InvalidConfig(format!(
221                    "{name} must be finite and not negative"
222                )));
223            }
224        }
225        self.validate_transport()?;
226        if self.socket_connection_setup_timeout_max < self.socket_connection_setup_timeout
227            || self.retry_backoff_max < self.retry_backoff
228        {
229            return Err(AdminError::InvalidConfig(
230                "maximum delay is below initial delay".into(),
231            ));
232        }
233        let api_timeout = default_api_timeout(self.request_timeout, self.default_api_timeout)?;
234        let client_id = match self.client_id {
235            Some(id) if !id.is_empty() => id,
236            _ => generate_client_id(),
237        };
238        let connection = ConnectionOptions {
239            client_id,
240            dns_timeout: self.dns_timeout,
241            client_dns_lookup: self.client_dns_lookup,
242            reconnect_backoff: self.reconnect_backoff,
243            reconnect_backoff_max: self.reconnect_backoff_max,
244            send_buffer: self.send_buffer,
245            receive_buffer: self.receive_buffer,
246            dispatch_queue_capacity: self.dispatch_queue_capacity,
247            frame_max: self.frame_max,
248            socket_connection_setup_timeout: self.socket_connection_setup_timeout,
249            socket_connection_setup_timeout_max: self.socket_connection_setup_timeout_max,
250            connections_max_idle: self.connections_max_idle,
251            request_timeout: self.request_timeout,
252            security: self.security.map(Box::new),
253            ..ConnectionOptions::default()
254        };
255        connection.validate().map_err(AdminError::InvalidConfig)?;
256        Ok(ResolvedAdminConfig {
257            connection,
258            retry: RetryPolicy {
259                timeout: api_timeout.to_std(),
260                initial_backoff: self.retry_backoff.to_std(),
261                max_backoff: self.retry_backoff_max.to_std(),
262                jitter: RETRY_BACKOFF_JITTER,
263                max_retries: self.retries,
264            },
265            metadata_recovery_strategy: self.metadata_recovery_strategy,
266            metadata_recovery_rebootstrap_trigger: self.metadata_recovery_rebootstrap_trigger,
267            enable_metrics_push: self.enable_metrics_push,
268        })
269    }
270}
271
272/// A new `adminclient-<n>` id.
273fn generate_client_id() -> String {
274    format!(
275        "adminclient-{}",
276        ADMIN_CLIENT_ID_SEQUENCE.fetch_add(1, Ordering::Relaxed)
277    )
278}
279
280/// The call deadline, as Kafka's `configureDefaultApiTimeoutMs` gives it: an
281/// explicit value below the request timeout is an error, and the default
282/// rises to the request timeout.
283fn default_api_timeout(request_timeout: Time, explicit: Option<Time>) -> Result<Time, AdminError> {
284    match explicit {
285        Some(api_timeout) if api_timeout < request_timeout => Err(AdminError::InvalidConfig(
286            "default_api_timeout must be no smaller than request_timeout".to_owned(),
287        )),
288        Some(api_timeout) => Ok(api_timeout),
289        None if DEFAULT_API_TIMEOUT < request_timeout => {
290            tracing::warn!(
291                request_timeout_ms = request_timeout.millis_i64(),
292                "overriding the default of default.api.timeout.ms (60000) with the configured \
293                 request timeout"
294            );
295            Ok(request_timeout)
296        }
297        None => Ok(DEFAULT_API_TIMEOUT),
298    }
299}
300
301/// The retry policy of a client built from bare [`ConnectionOptions`]:
302/// Kafka's defaults, with the call deadline raised to the request timeout.
303pub fn retry_for_connection_options(options: &ConnectionOptions) -> RetryPolicy {
304    let timeout = if options.request_timeout > DEFAULT_API_TIMEOUT {
305        options.request_timeout
306    } else {
307        DEFAULT_API_TIMEOUT
308    };
309    RetryPolicy {
310        timeout: timeout.to_std(),
311        initial_backoff: DEFAULT_RETRY_BACKOFF.to_std(),
312        max_backoff: DEFAULT_RETRY_BACKOFF_MAX.to_std(),
313        jitter: RETRY_BACKOFF_JITTER,
314        max_retries: u32::MAX,
315    }
316}
317
318#[cfg(test)]
319mod tests {
320    use std::{
321        sync::{
322            Arc, Mutex,
323            atomic::{AtomicUsize, Ordering},
324        },
325        time::Duration,
326    };
327
328    use assert2::{assert, check};
329    use bytes::{Buf, BytesMut};
330    use krabka_client_core::MockBroker;
331    use krabka_protocol::{
332        Encode,
333        owned::{
334            api_versions_request,
335            api_versions_response::{ApiVersion, ApiVersionsResponse},
336            unregister_broker_request,
337            unregister_broker_response::UnregisterBrokerResponse,
338        },
339    };
340    use tokio::time::Instant;
341
342    use super::*;
343    use crate::{AdminClient, AdminClientLike as _};
344
345    /// The settings that a test compares: client id prefix or value, request
346    /// timeout, connection setup timeouts, idle time, and the retry policy.
347    #[derive(Debug, PartialEq)]
348    struct Observed {
349        client_id: String,
350        request_timeout: Time,
351        setup_timeouts: (Time, Time),
352        connections_max_idle: Time,
353        retry: RetryPolicy,
354    }
355
356    fn observe(config: AdminClientConfig) -> Result<Observed, String> {
357        config
358            .resolve()
359            .map(|resolved| Observed {
360                // A generated id ends with a process-wide sequence number.
361                client_id: if resolved.connection.client_id.starts_with("adminclient-") {
362                    "adminclient-<n>".to_owned()
363                } else {
364                    resolved.connection.client_id
365                },
366                request_timeout: resolved.connection.request_timeout,
367                setup_timeouts: (
368                    resolved.connection.socket_connection_setup_timeout,
369                    resolved.connection.socket_connection_setup_timeout_max,
370                ),
371                connections_max_idle: resolved.connection.connections_max_idle,
372                retry: resolved.retry,
373            })
374            .map_err(|error| error.to_string())
375    }
376
377    fn retry(timeout: Duration) -> RetryPolicy {
378        RetryPolicy {
379            timeout,
380            initial_backoff: Duration::from_millis(100),
381            max_backoff: Duration::from_secs(1),
382            jitter: 0.2,
383            max_retries: u32::MAX,
384        }
385    }
386
387    /// Kafka's `AdminClientConfig` defaults, `generateClientId`, and
388    /// `configureDefaultApiTimeoutMs`.
389    #[test]
390    fn settings_resolve_as_kafka_admin_client_config() {
391        let defaults = || Observed {
392            client_id: "adminclient-<n>".into(),
393            request_timeout: secs(30),
394            setup_timeouts: (secs(10), secs(30)),
395            connections_max_idle: krabka_units::minutes(5),
396            retry: retry(Duration::from_mins(1)),
397        };
398        for (name, config, expected) in [
399            ("defaults", AdminClientConfig::default(), Ok(defaults())),
400            (
401                "request timeout 90 s raises the default api timeout",
402                AdminClientConfig {
403                    request_timeout: secs(90),
404                    ..AdminClientConfig::default()
405                },
406                Ok(Observed {
407                    request_timeout: secs(90),
408                    retry: retry(Duration::from_secs(90)),
409                    ..defaults()
410                }),
411            ),
412            (
413                "api timeout below the request timeout is an error",
414                AdminClientConfig {
415                    request_timeout: secs(90),
416                    default_api_timeout: Some(secs(60)),
417                    ..AdminClientConfig::default()
418                },
419                Err(
420                    "invalid admin client configuration: default_api_timeout must be no smaller \
421                     than request_timeout"
422                        .to_owned(),
423                ),
424            ),
425            (
426                "api timeout equal to the request timeout",
427                AdminClientConfig {
428                    default_api_timeout: Some(secs(30)),
429                    ..AdminClientConfig::default()
430                },
431                Ok(Observed {
432                    retry: retry(Duration::from_secs(30)),
433                    ..defaults()
434                }),
435            ),
436            (
437                "explicit client id",
438                AdminClientConfig {
439                    client_id: Some("tool-a".into()),
440                    ..AdminClientConfig::default()
441                },
442                Ok(Observed {
443                    client_id: "tool-a".into(),
444                    ..defaults()
445                }),
446            ),
447            (
448                "empty client id is generated",
449                AdminClientConfig {
450                    client_id: Some(String::new()),
451                    ..AdminClientConfig::default()
452                },
453                Ok(defaults()),
454            ),
455            (
456                "retries, backoff and setup timeout",
457                AdminClientConfig {
458                    retries: 3,
459                    retry_backoff: millis(50),
460                    retry_backoff_max: secs(2),
461                    socket_connection_setup_timeout: secs(5),
462                    socket_connection_setup_timeout_max: secs(20),
463                    connections_max_idle: secs(60),
464                    ..AdminClientConfig::default()
465                },
466                Ok(Observed {
467                    setup_timeouts: (secs(5), secs(20)),
468                    connections_max_idle: secs(60),
469                    retry: RetryPolicy {
470                        initial_backoff: Duration::from_millis(50),
471                        max_backoff: Duration::from_secs(2),
472                        max_retries: 3,
473                        ..retry(Duration::from_mins(1))
474                    },
475                    ..defaults()
476                }),
477            ),
478            (
479                "zero request timeout is an error",
480                AdminClientConfig {
481                    request_timeout: Time::ZERO,
482                    ..AdminClientConfig::default()
483                },
484                Err(
485                    "invalid admin client configuration: request_timeout must be positive and \
486                     finite"
487                        .to_owned(),
488                ),
489            ),
490            (
491                "infinite request timeout is an error",
492                AdminClientConfig {
493                    request_timeout: Time::from_secs_f64(f64::INFINITY),
494                    ..AdminClientConfig::default()
495                },
496                Err(
497                    "invalid admin client configuration: request_timeout must be positive and \
498                     finite"
499                        .to_owned(),
500                ),
501            ),
502            (
503                "NaN api timeout is an error",
504                AdminClientConfig {
505                    default_api_timeout: Some(Time::from_secs_f64(f64::NAN)),
506                    ..AdminClientConfig::default()
507                },
508                Err(
509                    "invalid admin client configuration: default_api_timeout must be positive \
510                     and finite"
511                        .to_owned(),
512                ),
513            ),
514            (
515                "NaN retry backoff is an error",
516                AdminClientConfig {
517                    retry_backoff: Time::from_secs_f64(f64::NAN),
518                    ..AdminClientConfig::default()
519                },
520                Err(
521                    "invalid admin client configuration: retry_backoff must be finite and not \
522                     negative"
523                        .to_owned(),
524                ),
525            ),
526        ] {
527            check!(observe(config) == expected, "{name}");
528        }
529    }
530
531    #[test]
532    fn generated_client_ids_follow_a_process_wide_sequence() {
533        let id = || {
534            AdminClientConfig::default()
535                .resolve()
536                .expect("defaults resolve")
537                .connection
538                .client_id
539        };
540        let number = |id: String| {
541            id.strip_prefix("adminclient-")
542                .and_then(|number| number.parse::<u32>().ok())
543                .expect("adminclient-<n>")
544        };
545        let first = number(id());
546        let second = number(id());
547        assert!(second > first);
548        assert!(first >= 1);
549    }
550
551    /// A broker that answers every `UnregisterBroker` with
552    /// `REQUEST_TIMED_OUT`, and records the client id of each request.
553    async fn timing_out_broker(
554        requests: Arc<AtomicUsize>,
555        client_ids: Arc<Mutex<Vec<String>>>,
556    ) -> MockBroker {
557        MockBroker::start(move |api_key, version, _, request| {
558            let mut header = request;
559            let client_id_len = usize::try_from(header.get_i16()).expect("client id length");
560            client_ids
561                .lock()
562                .expect("client ids lock")
563                .push(String::from_utf8_lossy(&header[..client_id_len]).into_owned());
564            let mut body = BytesMut::new();
565            match api_key {
566                api_versions_request::API_KEY => ApiVersionsResponse {
567                    api_keys: vec![
568                        ApiVersion {
569                            api_key: api_versions_request::API_KEY,
570                            ..Default::default()
571                        },
572                        ApiVersion {
573                            api_key: unregister_broker_request::API_KEY,
574                            ..Default::default()
575                        },
576                    ],
577                    ..Default::default()
578                }
579                .encode(&mut body, 0)
580                .unwrap(),
581                unregister_broker_request::API_KEY => {
582                    requests.fetch_add(1, Ordering::SeqCst);
583                    body.extend_from_slice(&[0]);
584                    UnregisterBrokerResponse {
585                        error_code: 7,
586                        ..Default::default()
587                    }
588                    .encode(&mut body, version)
589                    .unwrap();
590                }
591                _ => return None,
592            }
593            Some(body.to_vec())
594        })
595        .await
596        .unwrap()
597    }
598
599    /// A client built from `AdminClientConfig` sends its client id in each
600    /// request header, stops a call at `default_api_timeout`, and stops after
601    /// `retries` retries.
602    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
603    async fn a_client_uses_its_client_id_call_deadline_and_retries() {
604        for (name, config, expected_requests) in [
605            (
606                "no retries",
607                AdminClientConfig {
608                    client_id: Some("tool-a".into()),
609                    retries: 0,
610                    ..AdminClientConfig::default()
611                },
612                1..=1,
613            ),
614            (
615                "two retries",
616                AdminClientConfig {
617                    client_id: Some("tool-a".into()),
618                    retries: 2,
619                    retry_backoff: millis(1),
620                    retry_backoff_max: millis(1),
621                    ..AdminClientConfig::default()
622                },
623                3..=3,
624            ),
625            (
626                "the call deadline stops the retries",
627                AdminClientConfig {
628                    client_id: Some("tool-a".into()),
629                    request_timeout: millis(100),
630                    default_api_timeout: Some(millis(400)),
631                    retry_backoff: millis(100),
632                    retry_backoff_max: millis(100),
633                    ..AdminClientConfig::default()
634                },
635                2..=6,
636            ),
637        ] {
638            let requests = Arc::new(AtomicUsize::new(0));
639            let client_ids = Arc::new(Mutex::new(Vec::new()));
640            let broker = timing_out_broker(Arc::clone(&requests), Arc::clone(&client_ids)).await;
641            let mut admin = AdminClient::connect_with_config(&[broker.addr.to_string()], config)
642                .await
643                .expect("admin connects");
644
645            let started = Instant::now();
646            let result = admin.unregister_broker(1).await;
647            let elapsed = started.elapsed();
648
649            broker.stop();
650            let code = match result {
651                Err(crate::AdminError::Broker { code, .. }) => Some(code),
652                _ => None,
653            };
654            let requests = requests.load(Ordering::SeqCst);
655            let client_ids = client_ids.lock().expect("client ids lock").clone();
656            check!(code == Some(7), "{name}");
657            check!(
658                expected_requests.contains(&requests),
659                "{name}: {requests} requests"
660            );
661            check!(elapsed < Duration::from_secs(5), "{name}");
662            check!(
663                client_ids.iter().all(|id| id == "tool-a") && !client_ids.is_empty(),
664                "{name}: {client_ids:?}"
665            );
666        }
667    }
668}
669
670#[cfg(test)]
671mod transport_configuration_tests {
672    use std::time::Duration;
673
674    use assert2::assert;
675    use krabka_client_core::{ClientDnsLookup, ClientFrameMax, ConnectionDispatchQueueCapacity};
676    use krabka_units::{
677        convert::{ByteSizeExt as _, TimeExt as _},
678        kibibytes, mebibytes, millis,
679    };
680
681    use super::AdminClientConfig;
682
683    #[test]
684    fn admin_resolves_transport_resources_without_losing_retry_policy() {
685        let config = AdminClientConfig {
686            client_dns_lookup: ClientDnsLookup::ResolveCanonicalBootstrapServersOnly,
687            reconnect_backoff: millis(27),
688            reconnect_backoff_max: millis(127),
689            send_buffer: None,
690            receive_buffer: Some(kibibytes(19)),
691            dispatch_queue_capacity: ConnectionDispatchQueueCapacity::new(7).expect("capacity"),
692            frame_max: ClientFrameMax::try_from(mebibytes(2)).expect("frame"),
693            retry_backoff: millis(37),
694            ..Default::default()
695        };
696        let resolved = config.resolve().expect("valid config");
697        let options = resolved.connection;
698        assert!(
699            (
700                options.client_dns_lookup,
701                options.reconnect_backoff,
702                options.reconnect_backoff_max,
703                options.send_buffer,
704                options.receive_buffer,
705                options.dispatch_queue_capacity.get(),
706                options.frame_max.size()
707            ) == (
708                ClientDnsLookup::ResolveCanonicalBootstrapServersOnly,
709                millis(27),
710                millis(127),
711                None,
712                Some(kibibytes(19)),
713                7,
714                mebibytes(2)
715            )
716        );
717        assert!(resolved.retry.initial_backoff == Duration::from_millis(37));
718        let invalid = AdminClientConfig {
719            reconnect_backoff: millis(2000),
720            ..Default::default()
721        };
722        assert!(invalid.resolve().is_err());
723    }
724
725    #[test]
726    fn admin_rejects_invalid_transport_bounds_and_socket_sizes() {
727        let invalid_backoff = AdminClientConfig {
728            reconnect_backoff: krabka_units::Time::from_secs_f64(f64::NAN),
729            ..Default::default()
730        };
731        assert!(invalid_backoff.resolve().is_err());
732
733        let fractional_socket_buffer = AdminClientConfig {
734            send_buffer: Some(krabka_units::ByteSize::from_bytes_f64(0.5)),
735            ..Default::default()
736        };
737        assert!(fractional_socket_buffer.resolve().is_err());
738
739        let inverted_setup_bounds = AdminClientConfig {
740            socket_connection_setup_timeout_max: krabka_units::millis(1),
741            ..Default::default()
742        };
743        assert!(inverted_setup_bounds.resolve().is_err());
744    }
745}
krabka_client_admin/config.rs

krabka_client_admin/
config.rs
