Skip to content

Commit 9b70a52

Browse files
committed
Capture errors from connecting to broker
1 parent 1eba7e3 commit 9b70a52

File tree

2 files changed

+27
-5
lines changed

2 files changed

+27
-5
lines changed

src/connection.rs

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use async_trait::async_trait;
22
use rand::prelude::*;
3+
use std::fmt::Display;
34
use std::ops::ControlFlow;
45
use std::sync::Arc;
56
use thiserror::Error;
@@ -53,6 +54,23 @@ pub enum Error {
5354

5455
pub type Result<T, E = Error> = std::result::Result<T, E>;
5556

57+
#[derive(Debug, Error)]
58+
pub struct MultiError(Vec<Box<dyn std::error::Error + Send + Sync>>);
59+
60+
impl Display for MultiError {
61+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62+
let mut needs_comma = false;
63+
for err in &self.0 {
64+
if needs_comma {
65+
write!(f, ", ")?;
66+
}
67+
needs_comma = true;
68+
write!(f, "{err}")?;
69+
}
70+
Ok(())
71+
}
72+
}
73+
5674
/// How to connect to a `Transport`
5775
#[async_trait]
5876
trait ConnectionHandler {
@@ -470,6 +488,7 @@ where
470488
let mut backoff = Backoff::new(backoff_config);
471489
backoff
472490
.retry_with_backoff("broker_connect", || async {
491+
let mut errors = Vec::<Box<dyn std::error::Error + Send + Sync>>::new();
473492
for broker in &brokers {
474493
let conn = broker
475494
.connect(
@@ -485,16 +504,14 @@ where
485504
Ok(transport) => transport,
486505
Err(e) => {
487506
warn!(%e, "Failed to connect to broker");
507+
errors.push(Box::new(e));
488508
continue;
489509
}
490510
};
491511

492512
return ControlFlow::Break(connection);
493513
}
494-
495-
let err = Box::<dyn std::error::Error + Send + Sync>::from(
496-
"Failed to connect to any broker, backing off".to_string(),
497-
);
514+
let err = Box::<dyn std::error::Error + Send + Sync>::from(MultiError(errors));
498515
let err: Arc<dyn std::error::Error + Send + Sync> = err.into();
499516
ControlFlow::Continue(ErrorOrThrottle::Error(err))
500517
})

tests/client.rs

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -721,7 +721,12 @@ async fn test_client_backoff_terminates() {
721721

722722
match client_builder.build().await {
723723
Err(rskafka::client::error::Error::Connection(e)) => {
724-
assert_eq!(e.to_string(), "all retries failed: Retry exceeded deadline");
724+
// Error can be slightly different depending on the exact underlying error.
725+
assert!(e.to_string().starts_with(concat!(
726+
"all retries failed: Retry exceeded deadline. ",
727+
"Source: Failed to connect to any broker, backing off. ",
728+
"Errors: error connecting to broker \"localhost:9000\""
729+
)));
725730
}
726731
_ => {
727732
unreachable!();

0 commit comments

Comments
 (0)