Remove Kafka support and make stream bus MQTT-only

Author: argahsuknesib

Cargo.toml

@@ -27,9 +27,6 @@ tokio-tungstenite = "0.21"
 reqwest = { version = "0.11", features = ["json"] }
 futures-util = "0.3"
 
-[target.'cfg(not(windows))'.dependencies]
-rdkafka = "0.38.0"
-
 [lib]
 name = "janus"
 path = "src/lib.rs"

src/bin/stream_bus_cli.rs

@@ -1,14 +1,13 @@
 //! Stream Bus CLI - Command Line tool for the Stream Bus to publish the data to a broker and storage.
 //!
 //! Usage:
-//!   stream-bus-cli --input data/sensors.nq --broker kafka --topics sensors --rate 64
 //!   stream-bus-cli --input data/sensors.nq --broker mqtt --topics sensors --rate 64 --loop-file
 //!   stream-bus-cli --input data/sensors.nq --broker none --rate 0
 
 use clap::Parser;
 use janus::storage::segmented_storage::StreamingSegmentedStorage;
 use janus::storage::util::StreamingConfig;
-use janus::stream_bus::{BrokerType, KafkaConfig, MqttConfig, StreamBus, StreamBusConfig};
+use janus::stream_bus::{BrokerType, MqttConfig, StreamBus, StreamBusConfig};
 use std::sync::Arc;
 
 #[derive(Parser, Debug)]
@@ -19,8 +18,8 @@ struct Args {
     #[arg(short, long)]
     input: String,
 
-    /// Broker type: kafka, mqtt, or none
-    #[arg(short, long, default_value = "kafka")]
+    /// Broker type: mqtt or none
+    #[arg(short, long, default_value = "mqtt")]
     broker: String,
 
     /// Topics to publish to (comma-separated)
@@ -39,10 +38,6 @@ struct Args {
     #[arg(long)]
     add_timestamps: bool,
 
-    /// Kafka bootstrap servers
-    #[arg(long, default_value = "localhost:9092")]
-    kafka_servers: String,
-
     /// MQTT host
     #[arg(long, default_value = "localhost")]
     mqtt_host: String,
@@ -76,12 +71,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let storage = Arc::new(storage);
 
     let broker_type = match args.broker.to_lowercase().as_str() {
-        "kafka" => BrokerType::Kafka,
         "mqtt" => BrokerType::Mqtt,
         "none" => BrokerType::None,
         _ => {
             eprintln!("Error: Unknown broker type: {}", args.broker);
-            eprintln!("Valid options: kafka, mqtt, none");
+            eprintln!("Valid options: mqtt, none");
             std::process::exit(1);
         }
     };
@@ -95,12 +89,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         rate_of_publishing: args.rate,
         loop_file: args.loop_file,
         add_timestamps: args.add_timestamps,
-        kafka_config: match broker_type {
-            BrokerType::Kafka => {
-                Some(KafkaConfig { bootstrap_servers: args.kafka_servers, ..Default::default() })
-            }
-            _ => None,
-        },
         mqtt_config: match broker_type {
             BrokerType::Mqtt => Some(MqttConfig {
                 host: args.mqtt_host,

src/http/server.rs

@@ -9,7 +9,7 @@ use crate::{
     parsing::rdf_parser,
     registry::query_registry::{QueryId, QueryRegistry},
     storage::segmented_storage::StreamingSegmentedStorage,
-    stream_bus::{BrokerType, KafkaConfig, MqttConfig, StreamBus, StreamBusConfig},
+    stream_bus::{BrokerType, MqttConfig, StreamBus, StreamBusConfig},
 };
 use axum::{
     extract::{
@@ -93,7 +93,6 @@ pub struct StartReplayRequest {
     pub loop_file: bool,
     #[serde(default = "default_true")]
     pub add_timestamps: bool,
-    pub kafka_config: Option<KafkaConfigDto>,
     pub mqtt_config: Option<MqttConfigDto>,
 }
 
@@ -113,13 +112,6 @@ fn default_true() -> bool {
     true
 }
 
-#[derive(Debug, Deserialize)]
-pub struct KafkaConfigDto {
-    pub bootstrap_servers: String,
-    pub client_id: String,
-    pub message_timeout_ms: String,
-}
-
 #[derive(Debug, Deserialize)]
 pub struct MqttConfigDto {
     pub host: String,
@@ -418,24 +410,17 @@ async fn start_replay(
 
     // Parse broker type
     let broker_type = match payload.broker_type.to_lowercase().as_str() {
-        "kafka" => BrokerType::Kafka,
         "mqtt" => BrokerType::Mqtt,
         "none" => BrokerType::None,
         _ => {
             return Err(ApiError::BadRequest(format!(
-                "Invalid broker type: {}. Use 'kafka', 'mqtt', or 'none'",
+                "Invalid broker type: {}. Use 'mqtt' or 'none'",
                 payload.broker_type
             )))
         }
     };
 
     // Convert configs
-    let kafka_config = payload.kafka_config.map(|cfg| KafkaConfig {
-        bootstrap_servers: cfg.bootstrap_servers,
-        client_id: cfg.client_id,
-        message_timeout_ms: cfg.message_timeout_ms,
-    });
-
     let mqtt_config = payload.mqtt_config.map(|cfg| MqttConfig {
         host: cfg.host,
         port: cfg.port,
@@ -450,7 +435,6 @@ async fn start_replay(
         rate_of_publishing: payload.rate_of_publishing,
         loop_file: payload.loop_file,
         add_timestamps: payload.add_timestamps,
-        kafka_config,
         mqtt_config,
     };
 

src/sources/kafka_adapter.rs

@@ -1,5 +1,3 @@
-#[cfg(not(windows))]
-pub mod kafka_adapter;
 pub mod mqtt_adapter;
 pub mod stream_ingestion_pipeline;
 pub mod stream_source;

src/sources/mod.rs

@@ -13,7 +13,7 @@ impl StreamIngestionPipeline {
         StreamIngestionPipeline { storage, sources: Vec::new() }
     }
 
-    /// Adding the source for the stream ingestion pipeline (which can be MQTT, Kafka, etc.)
+    /// Adding a source for the stream ingestion pipeline (for example MQTT).
     pub fn add_source(&mut self, source: Box<dyn StreamSource>) {
         self.sources.push(source);
     }

src/sources/stream_ingestion_pipeline.rs

@@ -1,6 +1,5 @@
 pub mod stream_bus;
 
 pub use stream_bus::{
-    BrokerType, KafkaConfig, MqttConfig, StreamBus, StreamBusConfig, StreamBusError,
-    StreamBusMetrics,
+    BrokerType, MqttConfig, StreamBus, StreamBusConfig, StreamBusError, StreamBusMetrics,
 };