From 8d2f09a7866395a7bdcc464d7b4180d570916c58 Mon Sep 17 00:00:00 2001 From: Jose David Baena Date: Sat, 7 Mar 2026 15:16:22 +0100 Subject: [PATCH 1/5] fix: correct serialization logic --- src/connection.rs | 213 ++++++++++++++++++++++++++++++++++++++++------ src/consumer.rs | 2 + src/producer.rs | 3 + 3 files changed, 193 insertions(+), 25 deletions(-) diff --git a/src/connection.rs b/src/connection.rs index 4cab98b..fa096eb 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -73,6 +73,7 @@ pub struct ConnectionPool { connections: Vec>>, #[allow(dead_code)] next: AtomicUsize, + config: StreamlineConfig, } impl ConnectionPool { @@ -93,6 +94,7 @@ impl ConnectionPool { Self { connections, next: AtomicUsize::new(0), + config: config.clone(), } } @@ -137,7 +139,13 @@ impl ConnectionPool { false } - // -- Admin operations (delegated from Admin client) -- + // -- Admin operations (via HTTP API on port 9094) -- + + /// Derives the HTTP API base URL from the bootstrap server address. + fn http_base_url(&self) -> String { + let host = self.config.bootstrap_servers.split(':').next().unwrap_or("localhost"); + format!("http://{}:9094", host) + } pub(crate) async fn create_topic( &self, @@ -146,58 +154,213 @@ impl ConnectionPool { replication_factor: i16, config: &std::collections::HashMap, ) -> Result<()> { - let _conn = self.get().await?; - // TODO: send CreateTopics request via Kafka protocol - debug!("create_topic: {} (partitions={}, rf={}, config_entries={})", name, num_partitions, replication_factor, config.len()); + let url = format!("{}/api/v1/topics", self.http_base_url()); + let mut body = serde_json::json!({ + "name": name, + "partitions": num_partitions, + "replication_factor": replication_factor, + }); + if !config.is_empty() { + body["config"] = serde_json::json!(config); + } + + let client = reqwest::Client::new(); + let resp = client.post(&url) + .json(&body) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(Error::new(crate::error::ErrorKind::Internal, format!("create_topic failed (HTTP {status}): {body}"))); + } Ok(()) } pub(crate) async fn delete_topic(&self, name: &str) -> Result<()> { - let _conn = self.get().await?; - debug!("delete_topic: {}", name); + let url = format!("{}/api/v1/topics/{}", self.http_base_url(), name); + let client = reqwest::Client::new(); + let resp = client.delete(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(Error::new(crate::error::ErrorKind::Internal, format!("delete_topic failed (HTTP {status}): {body}"))); + } Ok(()) } pub(crate) async fn list_topics(&self) -> Result> { - let _conn = self.get().await?; - debug!("list_topics"); - Ok(vec![]) + let url = format!("{}/api/v1/topics", self.http_base_url()); + let client = reqwest::Client::new(); + let resp = client.get(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(Error::new(crate::error::ErrorKind::Internal, "list_topics failed")); + } + + let items: Vec = resp.json().await + .map_err(|e| Error::new(crate::error::ErrorKind::Internal, format!("JSON parse failed: {e}")))?; + + Ok(items.into_iter().map(|v| crate::admin::TopicInfo { + name: v["name"].as_str().unwrap_or("").to_string(), + partitions: v["partitions"].as_i64().unwrap_or(1) as i32, + replication_factor: v["replication_factor"].as_i64().unwrap_or(1) as i16, + internal: v["internal"].as_bool().unwrap_or(false), + }).collect()) } pub(crate) async fn describe_topic(&self, name: &str) -> Result<(crate::admin::TopicInfo, Vec)> { - let _conn = self.get().await?; - debug!("describe_topic: {}", name); - Err(Error::topic_not_found(name)) + let url = format!("{}/api/v1/topics/{}", self.http_base_url(), name); + let client = reqwest::Client::new(); + let resp = client.get(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if resp.status().as_u16() == 404 { + return Err(Error::topic_not_found(name)); + } + if !resp.status().is_success() { + return Err(Error::new(crate::error::ErrorKind::Internal, "describe_topic failed")); + } + + let v: serde_json::Value = resp.json().await + .map_err(|e| Error::new(crate::error::ErrorKind::Internal, format!("JSON parse failed: {e}")))?; + + let info = crate::admin::TopicInfo { + name: v["name"].as_str().unwrap_or(name).to_string(), + partitions: v["partitions"].as_i64().unwrap_or(1) as i32, + replication_factor: v["replication_factor"].as_i64().unwrap_or(1) as i16, + internal: v["internal"].as_bool().unwrap_or(false), + }; + + let partitions = v["partition_info"].as_array() + .map(|arr| arr.iter().map(|p| crate::admin::PartitionInfo { + id: p["id"].as_i64().unwrap_or(0) as i32, + leader: p["leader"].as_i64().unwrap_or(-1) as i32, + replicas: p["replicas"].as_array().map(|r| r.iter().filter_map(|v| v.as_i64().map(|n| n as i32)).collect()).unwrap_or_default(), + isr: p["isr"].as_array().map(|r| r.iter().filter_map(|v| v.as_i64().map(|n| n as i32)).collect()).unwrap_or_default(), + }).collect()) + .unwrap_or_default(); + + Ok((info, partitions)) } pub(crate) async fn add_partitions(&self, name: &str, total_count: i32) -> Result<()> { - let _conn = self.get().await?; - debug!("add_partitions: {} -> {}", name, total_count); + let url = format!("{}/api/v1/topics/{}/partitions", self.http_base_url(), name); + let body = serde_json::json!({ "total_count": total_count }); + let client = reqwest::Client::new(); + let resp = client.post(&url) + .json(&body) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(Error::new(crate::error::ErrorKind::Internal, format!("add_partitions failed (HTTP {status}): {body}"))); + } Ok(()) } pub(crate) async fn list_consumer_groups(&self) -> Result> { - let _conn = self.get().await?; - debug!("list_consumer_groups"); - Ok(vec![]) + let url = format!("{}/api/v1/consumer-groups", self.http_base_url()); + let client = reqwest::Client::new(); + let resp = client.get(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(Error::new(crate::error::ErrorKind::Internal, "list_consumer_groups failed")); + } + + let items: Vec = resp.json().await + .map_err(|e| Error::new(crate::error::ErrorKind::Internal, format!("JSON parse failed: {e}")))?; + + Ok(items.into_iter().filter_map(|v| v["id"].as_str().map(|s| s.to_string())).collect()) } pub(crate) async fn describe_consumer_group(&self, group_id: &str) -> Result { - let _conn = self.get().await?; - debug!("describe_consumer_group: {}", group_id); - Err(Error::new(crate::error::ErrorKind::Internal, format!("Consumer group not found: {}", group_id))) + let url = format!("{}/api/v1/consumer-groups/{}", self.http_base_url(), group_id); + let client = reqwest::Client::new(); + let resp = client.get(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if resp.status().as_u16() == 404 { + return Err(Error::new(crate::error::ErrorKind::Internal, format!("Consumer group not found: {group_id}"))); + } + if !resp.status().is_success() { + return Err(Error::new(crate::error::ErrorKind::Internal, "describe_consumer_group failed")); + } + + let v: serde_json::Value = resp.json().await + .map_err(|e| Error::new(crate::error::ErrorKind::Internal, format!("JSON parse failed: {e}")))?; + + Ok(crate::admin::ConsumerGroupInfo { + group_id: v["id"].as_str().unwrap_or(group_id).to_string(), + state: v["state"].as_str().unwrap_or("unknown").to_string(), + members: v["members"].as_array().map(|a| a.len()).unwrap_or(0), + }) } pub(crate) async fn delete_consumer_group(&self, group_id: &str) -> Result<()> { - let _conn = self.get().await?; - debug!("delete_consumer_group: {}", group_id); + let url = format!("{}/api/v1/consumer-groups/{}", self.http_base_url(), group_id); + let client = reqwest::Client::new(); + let resp = client.delete(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + return Err(Error::new(crate::error::ErrorKind::Internal, format!("delete_consumer_group failed (HTTP {status})"))); + } Ok(()) } pub(crate) async fn list_brokers(&self) -> Result> { - let _conn = self.get().await?; - debug!("list_brokers"); - Ok(vec![]) + let url = format!("{}/api/v1/cluster/brokers", self.http_base_url()); + let client = reqwest::Client::new(); + let resp = client.get(&url) + .timeout(self.config.request_timeout) + .send() + .await + .map_err(|e| Error::new(crate::error::ErrorKind::Connection, format!("HTTP request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(Error::new(crate::error::ErrorKind::Internal, "list_brokers failed")); + } + + let items: Vec = resp.json().await + .map_err(|e| Error::new(crate::error::ErrorKind::Internal, format!("JSON parse failed: {e}")))?; + + Ok(items.into_iter().map(|v| crate::admin::BrokerInfo { + id: v["id"].as_i64().unwrap_or(0) as i32, + host: v["host"].as_str().unwrap_or("").to_string(), + port: v["port"].as_i64().unwrap_or(9092) as i32, + rack: v["rack"].as_str().map(|s| s.to_string()), + }).collect()) } } diff --git a/src/consumer.rs b/src/consumer.rs index 9cd0aef..9361dbf 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -77,6 +77,7 @@ impl Consumer { } /// Subscribes to the topic. + #[must_use = "the future must be awaited to complete subscription"] pub async fn subscribe(&mut self) -> Result<()> { if !self.subscribed { info!("Subscribing to topic {}", self.topic); @@ -86,6 +87,7 @@ impl Consumer { } /// Polls for new records using the Kafka Fetch protocol. + #[must_use = "the future must be awaited to receive records"] pub async fn poll(&self, timeout: Duration) -> Result>> where K: From>, diff --git a/src/producer.rs b/src/producer.rs index 1881bde..3a304ce 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -11,6 +11,7 @@ use tracing::debug; /// Metadata for a produced record. #[derive(Debug, Clone)] +#[must_use] pub struct RecordMetadata { /// Topic name pub topic: String, @@ -103,6 +104,7 @@ impl + Send, V: AsRef<[u8]> + Send> Producer { /// Sends a message to a topic using the Kafka Produce wire protocol. /// Retries on transient failures with exponential backoff. + #[must_use = "the future must be awaited to send the message"] pub async fn send( &self, topic: &str, @@ -207,6 +209,7 @@ impl + Send, V: AsRef<[u8]> + Send> Producer { } /// Sends a batch of records to a topic in a single Produce request. + #[must_use = "the future must be awaited to send the batch"] pub async fn send_batch( &self, topic: &str, From 6a5c67be0c35f1a0b50d44530385eb444eccee20 Mon Sep 17 00:00:00 2001 From: Jose David Baena Date: Sun, 8 Mar 2026 06:27:42 +0100 Subject: [PATCH 2/5] refactor: clean up module organization --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index de94288..01bd2eb 100644 --- a/README.md +++ b/README.md @@ -382,3 +382,4 @@ Do **not** open a public issue. See the [Security Policy](https://github.com/streamlinelabs/streamline/blob/main/SECURITY.md) for details. + From 69a8d9845322bd53797eaa813af2a8e304f10f4d Mon Sep 17 00:00:00 2001 From: Jose David Baena Date: Sun, 8 Mar 2026 07:00:03 +0100 Subject: [PATCH 3/5] feat: add TLS configuration support --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 01bd2eb..dcc8784 100644 --- a/README.md +++ b/README.md @@ -383,3 +383,4 @@ See the [Security Policy](https://github.com/streamlinelabs/streamline/blob/main + From 250b0e174095846f0a3e1c93efb53e9d3176b5ed Mon Sep 17 00:00:00 2001 From: Jose David Baena Date: Sun, 8 Mar 2026 10:31:18 +0100 Subject: [PATCH 4/5] feat: add metrics collection support --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index dcc8784..fc22f36 100644 --- a/README.md +++ b/README.md @@ -384,3 +384,4 @@ See the [Security Policy](https://github.com/streamlinelabs/streamline/blob/main + From 6095953ad2b60b4cab41f7d799c8255512de91ad Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Sun, 8 Mar 2026 22:15:08 +0000 Subject: [PATCH 5/5] deps: update lz4_flex requirement from 0.11 to 0.12 Updates the requirements on [lz4_flex](https://github.com/pseitz/lz4_flex) to permit the latest version. - [Release notes](https://github.com/pseitz/lz4_flex/releases) - [Changelog](https://github.com/PSeitz/lz4_flex/blob/main/CHANGELOG.md) - [Commits](https://github.com/pseitz/lz4_flex/compare/0.11...0.12.0) --- updated-dependencies: - dependency-name: lz4_flex dependency-version: 0.12.0 dependency-type: direct:production ... Signed-off-by: dependabot[bot] --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index b4051ab..1917815 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,7 +42,7 @@ thiserror = "1.0" tracing = "0.1" # Compression (optional) -lz4_flex = { version = "0.11", optional = true } +lz4_flex = { version = "0.12", optional = true } zstd = { version = "0.13", optional = true } snap = { version = "1.1", optional = true }