Skip to content

Commit

Permalink
dekaf: Introduce transient error retry limit
Browse files Browse the repository at this point in the history
  • Loading branch information
jshearer committed Sep 24, 2024
1 parent 6ecc56f commit b3d445d
Showing 1 changed file with 5 additions and 3 deletions.
8 changes: 5 additions & 3 deletions crates/dekaf/src/read.rs
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ impl Read {
let mut buf = bytes::BytesMut::new();

let mut has_had_parsing_error = false;
let mut transient_errors = 0;

while records_bytes < target_bytes {
let read = match tokio::select! {
Expand Down Expand Up @@ -115,14 +116,15 @@ impl Read {
continue;
}
},
Err(err) if err.is_transient() => {
Err(err) if err.is_transient() && transient_errors < 5 => {
use rand::Rng;

tracing::warn!(%err, "Retrying transient read error");
transient_errors = transient_errors + 1;

tracing::warn!(error = ?err, "Retrying transient read error");
let delay = Duration::from_millis(rand::thread_rng().gen_range(300..2000));
tokio::time::sleep(delay).await;
// We can retry transient errors just by continuing to poll the stream
// TODO: We might have a counter here and give up after a few attempts
continue;
}
Err(err @ gazette::Error::Parsing { .. }) if !has_had_parsing_error => {
Expand Down

0 comments on commit b3d445d

Please sign in to comment.