feat(metrics): add KafkaMetricsCollector for Prometheus integration .

Implement KafkaMetricsCollector to collect and expose Kafka client and broker metrics. Refactor ParseableServer.init(..) and connectors::init(..).
parseablehq · Dec 23, 2024 · 9ce1031 · 9ce1031
1 parent aff48a2
commit 9ce1031
Show file tree

Hide file tree

Showing 11 changed files with 714 additions and 373 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -32,8 +32,6 @@ mime = "0.3.17"
 
 ### connectors dependencies
 rdkafka = { version = "0.37", features = ["cmake-build", "tracing", "sasl", "ssl", "libz-static"] }
-testcontainers = "0.23"
-testcontainers-modules = { version = "0.11", features = ["kafka"] }
 backon = "1.3"
 
 ### other dependencies

diff --git a/src/connectors/kafka/config.rs b/src/connectors/kafka/config.rs
@@ -479,7 +479,7 @@ impl Default for KafkaConfig {
             // Common configuration with standard broker port
             bootstrap_servers: "localhost:9092".to_string(),
             topics: vec![],
-            client_id: None, // Let Kafka generate a unique client ID if not specified
+            client_id: Some("parseable-connect".to_string()),
 
             // Component-specific configurations with production-ready defaults
             consumer: Some(ConsumerConfig::default()),
@@ -613,7 +613,10 @@ mod tests {
         );
 
         let rdkafka_config = config.consumer_config();
-        assert_eq!(rdkafka_config.get("group.id"), Some("test-group"));
+        assert_eq!(
+            rdkafka_config.get("group.id"),
+            Some("parseable-test-group-gi")
+        );
         assert_eq!(
             rdkafka_config.get("partition.assignment.strategy"),
             Some("cooperative-sticky")
@@ -631,7 +634,7 @@ mod tests {
         if let Some(producer) = config.producer {
             assert_eq!(producer.acks, "all");
             assert!(producer.enable_idempotence);
-            assert_eq!(producer.compression_type, "snappy");
+            assert_eq!(producer.compression_type, "lz4");
         }
     }
 

diff --git a/src/connectors/kafka/consumer.rs b/src/connectors/kafka/consumer.rs
@@ -47,8 +47,8 @@ impl KafkaStreams {
         shutdown_handle: Shutdown,
     ) -> anyhow::Result<KafkaStreams> {
         info!("Initializing KafkaStreams...");
+        let statistics = Arc::clone(&context.statistics);
         let consumer = KafkaStreams::create_consumer(context);
-        let statistics = Arc::new(std::sync::RwLock::new(Statistics::default()));
         info!("KafkaStreams initialized successfully.");
 
         Ok(Self {
@@ -120,15 +120,12 @@ impl KafkaStreams {
                     )
                     .await;
 
-                    match result {
-                        Err(e) => {
-                            error!(
-                                "Partitioned processing encountered a critical error: {:?}",
-                                e
-                            );
-                            break;
-                        }
-                        Ok(..) => {}
+                    if let Err(e) = result {
+                        error!(
+                            "Partitioned processing encountered a critical error: {:?}",
+                            e
+                        );
+                        break;
                     }
                 }
             });
@@ -169,7 +166,7 @@ impl KafkaStreams {
         let recv_fn = || consumer.recv();
 
         recv_fn
-            .retry(retry_policy.clone())
+            .retry(*retry_policy)
             .sleep(tokio::time::sleep)
             .notify(|err, dur| {
                 tracing::warn!(