dekaf: Temporarily implement and use `flow_client::Client::with_fresh…

…_gazette_client()`
estuary · Oct 7, 2024 · cd09a83 · cd09a83
1 parent 1896afc
commit cd09a83
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 4 deletions.
diff --git a/crates/dekaf/src/lib.rs b/crates/dekaf/src/lib.rs
@@ -72,7 +72,8 @@ impl App {
         let client = self
             .client_base
             .clone()
-            .with_creds(Some(access), Some(refresh));
+            .with_creds(Some(access), Some(refresh))
+            .with_fresh_gazette_client();
 
         let claims = flow_client::client::client_claims(&client)?;
 

diff --git a/crates/flow-client/src/client.rs b/crates/flow-client/src/client.rs
@@ -68,7 +68,18 @@ impl Client {
         user_access_token: Option<String>,
         user_refresh_token: Option<RefreshToken>,
     ) -> Self {
-        // Test whether a fresh set of Gazette clients solves the timeout issue
+        Self {
+            user_access_token: user_access_token.or(self.user_access_token),
+            user_refresh_token: user_refresh_token.or(self.user_refresh_token),
+            ..self
+        }
+    }
+
+    /// Build a fresh `gazette::journal::Client` and `gazette::shard::Client`
+    /// There is a bug that causes these clients to hang under heavy/varied load,
+    /// so until that bug is found+fixed, this is the work-around.
+    #[deprecated]
+    pub fn with_fresh_gazette_client(self) -> Self {
         let router = gazette::Router::new("local");
 
         let journal_client = gazette::journal::Client::new(
@@ -82,8 +93,6 @@ impl Client {
             router.clone(),
         );
         Self {
-            user_access_token: user_access_token.or(self.user_access_token),
-            user_refresh_token: user_refresh_token.or(self.user_refresh_token),
             journal_client,
             shard_client,
             ..self