fix: build fix for Kafka (#1079)

 Signed-off-by: parmesant <anant.v09@proton.me> Co-authored-by: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Co-authored-by: Devdutt Shenoi <devdutt@outlook.in>

Author: 3 people

Cargo.toml

@@ -67,7 +67,6 @@ once_cell = "1.17.1"
 opentelemetry-proto = {git = "https://github.com/parseablehq/opentelemetry-rust", branch="fix-metrics-u64-serialization"}
 prometheus = { version = "0.13", features = ["process"] }
 rand = "0.8.5"
-rdkafka = { version = "0.36.2", default-features = false, features = ["tokio"] }
 regex = "1.7.3"
 relative-path = { version = "1.7", features = ["serde"] }
 reqwest = { version = "0.11.27", default-features = false, features = [
@@ -136,3 +135,7 @@ debug = []
 inherits = "release"
 lto = "fat"
 codegen-units = 1
+
+# adding rdkafka here because, for unsupported platforms, cargo skips other deps which come after this
+[target.'cfg(all(target_os = "linux", target_arch = "x86_64"))'.dependencies]
+rdkafka = { version = "0.36.2", default-features = false, features = ["tokio"] }

src/cli.rs

@@ -22,11 +22,16 @@ use std::path::PathBuf;
 use url::Url;
 
 use crate::{
- kafka::SslProtocol,
  oidc::{self, OpenidConfig},
  option::{validation, Compression, Mode},
 };
 
+#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+use crate::kafka::SslProtocol as KafkaSslProtocol;
+
+#[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
+use std::string::String as KafkaSslProtocol;
+
 #[derive(Debug, Default)]
 pub struct Cli {
  /// The location of TLS Cert file
@@ -107,7 +112,7 @@ pub struct Cli {
  pub kafka_host: Option<String>,
  pub kafka_group: Option<String>,
  pub kafka_client_id: Option<String>,
- pub kafka_security_protocol: Option<SslProtocol>,
+ pub kafka_security_protocol: Option<KafkaSslProtocol>,
  pub kafka_partitions: Option<String>,
 
  // Audit Logging env vars
@@ -502,16 +507,17 @@ impl FromArgMatches for Cli {
  }
 
  fn update_from_arg_matches(&mut self, m: &clap::ArgMatches) -> Result<(), clap::Error> {
- self.kafka_topics = m.get_one::<String>(Self::KAFKA_TOPICS).cloned();
- self.kafka_security_protocol = m
- .get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL)
- .cloned();
- self.kafka_group = m.get_one::<String>(Self::KAFKA_GROUP).cloned();
- self.kafka_client_id = m.get_one::<String>(Self::KAFKA_CLIENT_ID).cloned();
- self.kafka_security_protocol = m
- .get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL)
- .cloned();
- self.kafka_partitions = m.get_one::<String>(Self::KAFKA_PARTITIONS).cloned();
+ #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+ {
+ self.kafka_topics = m.get_one::<String>(Self::KAFKA_TOPICS).cloned();
+ self.kafka_security_protocol = m
+ .get_one::<KafkaSslProtocol>(Self::KAFKA_SECURITY_PROTOCOL)
+ .cloned();
+ self.kafka_group = m.get_one::<String>(Self::KAFKA_GROUP).cloned();
+ self.kafka_client_id = m.get_one::<String>(Self::KAFKA_CLIENT_ID).cloned();
+ self.kafka_host = m.get_one::<String>(Self::KAFKA_HOST).cloned();
+ self.kafka_partitions = m.get_one::<String>(Self::KAFKA_PARTITIONS).cloned();
+ }
 
  self.audit_logger = m.get_one::<Url>(Self::AUDIT_LOGGER).cloned();
  self.audit_username = m.get_one::<String>(Self::AUDIT_USERNAME).cloned();

src/handlers/http/health_check.rs

@@ -58,7 +58,7 @@ pub async fn handle_signals(shutdown_signal: Arc<Mutex<Option<oneshot::Sender<()
  {
  tokio::select! {
  _ = ctrl_c() => {
- log::info!("Received SIGINT signal at Readiness Probe Handler");
+ info!("Received SIGINT signal at Readiness Probe Handler");
  shutdown(shutdown_signal).await;
  }
  }

src/lib.rs

@@ -27,6 +27,7 @@ pub mod correlation;
 mod event;
 pub mod handlers;
 pub mod hottier;
+#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
 pub mod kafka;
 mod livetail;
 mod metadata;

src/main.rs

@@ -17,12 +17,15 @@
  */
 
 use parseable::{
- banner, kafka,
+ banner,
  option::{Mode, CONFIG},
  rbac, storage, IngestServer, ParseableServer, QueryServer, Server,
 };
 use tracing_subscriber::EnvFilter;
 
+#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+use parseable::kafka;
+
 #[actix_web::main]
 async fn main() -> anyhow::Result<()> {
  tracing_subscriber::fmt()
@@ -46,6 +49,7 @@ async fn main() -> anyhow::Result<()> {
  // keep metadata info in mem
  metadata.set_global();
 
+ #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
  // load kafka server
  if CONFIG.parseable.mode != Mode::Query {
  tokio::task::spawn(kafka::setup_integration());

src/query/stream_schema_provider.rs

@@ -301,7 +301,9 @@ impl StandardTableProvider {
  #[cfg(windows)]
  {
  if CONFIG.storage_name.eq("drive") {
- file_path = object_store::path::Path::from_absolute_path(file_path).unwrap();
+ file_path = object_store::path::Path::from_absolute_path(file_path)
+ .unwrap()
+ .to_string();
  }
  }
  let pf = PartitionedFile::new(file_path, file.file_size);