dekaf: Random delay when retrying transient error

estuary · Sep 16, 2024 · f763d27 · f763d27
1 parent 8e7e279
commit f763d27
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 0 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/dekaf/Cargo.toml b/crates/dekaf/Cargo.toml
@@ -43,6 +43,7 @@ metrics-prometheus = { workspace = true }
 percent-encoding = { workspace = true }
 postgrest = { workspace = true }
 prometheus = { workspace = true }
+rand = { workspace = true }
 regex = { workspace = true }
 rsasl = { workspace = true }
 rustls = { workspace = true }

diff --git a/crates/dekaf/src/read.rs b/crates/dekaf/src/read.rs
@@ -4,6 +4,7 @@ use doc::AsNode;
 use futures::StreamExt;
 use gazette::journal::{ReadJsonLine, ReadJsonLines};
 use gazette::{broker, journal, uuid};
+use std::time::Duration;
 
 pub struct Read {
  /// Journal offset to be served by this Read.
@@ -112,7 +113,11 @@ impl Read {
  }
  },
  Err(err) if err.is_transient() => {
+ use rand::Rng;
+
  tracing::warn!(%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;