Fix logic that makes LagRegister forget groups

kafkesc · Feb 15, 2024 · ff32275 · ff32275
1 parent 386a24a
commit ff32275
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/src/consumer_groups/register.rs b/src/consumer_groups/register.rs
@@ -21,7 +21,7 @@ use crate::internals::Awaitable;
 use crate::kafka_types::GroupWithMembers;
 use crate::prometheus_metrics::LABEL_GROUP;
 
-const REMOVE_EXPIRED_INTERVAL: Duration = Duration::from_millis(500);
+const REMOVE_EXPIRED_INTERVAL: Duration = Duration::from_secs(1);
 
 const MET_TOT_NAME: &str = "consumer_groups_total";
 const MET_TOT_HELP: &str = "Consumer groups currently in the cluster";
@@ -193,9 +193,11 @@ impl ConsumerGroupsRegister {
                     },
                 }
 
-                info!(
-                    "Updated (Known) Consumer Groups: {}",
-                    known_groups_arc_clone.read().await.map.len()
+                let r_guard = known_groups_arc_clone.read().await;
+                trace!(
+                    "Updated (Known) Consumer Groups: {} (hash: {})",
+                    r_guard.map.len(),
+                    r_guard.hash,
                 );
             }
         });

diff --git a/src/lag_register/register.rs b/src/lag_register/register.rs
@@ -209,7 +209,7 @@ async fn process_consumer_groups(
     }
 
     // ... then, remove groups that are in `lag_register_groups` but are not known (anymore)
-    lag_register_groups.write().await.retain(|g, _| !known_groups.contains(g));
+    lag_register_groups.write().await.retain(|g, _| known_groups.contains(g));
 }
 
 async fn process_offset_commit(