Merge branch 'main' into correlation-backend

.github/workflows/integration-test.yaml

@@ -1,12 +1,11 @@
-name: Integration
-
+name: Integration Tests with Quest
 on:
   pull_request:
     paths-ignore:
-      - 'docs/**'
-      - 'helm/**'
-      - 'assets/**'
-      - '**.md'
+      - "docs/**"
+      - "helm/**"
+      - "assets/**"
+      - "**.md"
 
 jobs:
 

.github/workflows/lint.yaml

@@ -0,0 +1,45 @@
+name: Lint
+on:
+  pull_request:
+    paths-ignore:
+      - "docs/**"
+      - "helm/**"
+      - "assets/**"
+      - "**.md"
+  push:
+    branches:
+      - main
+
+jobs:
+
+  fmt:
+    name: Rust fmt check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+      - run: rustup component add rustfmt
+      - uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
+
+  clippy:
+    name: Cargo Clippy check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+      - run: rustup component add clippy
+      - uses: actions-rs/cargo@v1
+        with:
+          command: clippy
+          args: -- -D warnings

.github/workflows/unit-test.yaml

@@ -0,0 +1,26 @@
+name: Unit Tests
+on:
+  pull_request:
+    paths-ignore:
+      - "docs/**"
+      - "helm/**"
+      - "assets/**"
+      - "**.md"
+  push:
+    branches:
+      - main
+
+jobs:
+  unit-tests:
+    name: Unit tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test

src/cli.rs

@@ -526,7 +526,9 @@ impl FromArgMatches for Cli {
         self.trino_username = m.get_one::<String>(Self::TRINO_USER_NAME).cloned();
 
         self.kafka_topics = m.get_one::<String>(Self::KAFKA_TOPICS).cloned();
-        self.kafka_host = m.get_one::<String>(Self::KAFKA_HOST).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

src/kafka.rs

@@ -125,7 +125,7 @@ pub enum KafkaError {
 fn setup_consumer() -> Result<(StreamConsumer, Vec<String>), KafkaError> {
     if let Some(topics) = &CONFIG.parseable.kafka_topics {
         // topics can be a comma separated list of topics to subscribe to
-        let topics = topics.split(",").map(|v| v.to_owned()).collect_vec();
+        let topics = topics.split(',').map(|v| v.to_owned()).collect_vec();
 
         let host = if CONFIG.parseable.kafka_host.is_some() {
             CONFIG.parseable.kafka_host.as_ref()
@@ -162,8 +162,8 @@ fn setup_consumer() -> Result<(StreamConsumer, Vec<String>), KafkaError> {
             // partitions is a comma separated pairs of topic:partitions
             let mut topic_partition_pairs = Vec::new();
             let mut set = true;
-            for vals in vals_raw.split(",") {
-                let intermediate = vals.split(":").collect_vec();
+            for vals in vals_raw.split(',') {
+                let intermediate = vals.split(':').collect_vec();
                 if intermediate.len() != 2 {
                     warn!(
                         "Value for P_KAFKA_PARTITIONS is incorrect! Skipping setting partitions!"