Clean up code

quickwit-oss · Aug 11, 2023 · cf1965b · cf1965b
1 parent aa33d1e
commit cf1965b
Show file tree

Hide file tree

Showing 10 changed files with 127 additions and 122 deletions.
diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock
diff --git a/quickwit/quickwit-config/src/source_config/mod.rs b/quickwit/quickwit-config/src/source_config/mod.rs
@@ -85,29 +85,29 @@ impl SourceConfig {
     pub fn source_type(&self) -> &str {
         match self.source_params {
             SourceParams::File(_) => "file",
-            SourceParams::Kafka(_) => "kafka",
             SourceParams::GcpPubSub(_) => "gcp_pubsub",
-            SourceParams::Kinesis(_) => "kinesis",
-            SourceParams::Vec(_) => "vec",
-            SourceParams::Void(_) => "void",
             SourceParams::IngestApi => "ingest-api",
             SourceParams::IngestCli => "ingest-cli",
+            SourceParams::Kafka(_) => "kafka",
+            SourceParams::Kinesis(_) => "kinesis",
             SourceParams::Pulsar(_) => "pulsar",
+            SourceParams::Vec(_) => "vec",
+            SourceParams::Void(_) => "void",
         }
     }
 
     // TODO: Remove after source factory refactor.
     pub fn params(&self) -> JsonValue {
         match &self.source_params {
             SourceParams::File(params) => serde_json::to_value(params),
-            SourceParams::Kafka(params) => serde_json::to_value(params),
             SourceParams::GcpPubSub(params) => serde_json::to_value(params),
-            SourceParams::Kinesis(params) => serde_json::to_value(params),
-            SourceParams::Vec(params) => serde_json::to_value(params),
-            SourceParams::Void(params) => serde_json::to_value(params),
             SourceParams::IngestApi => serde_json::to_value(()),
             SourceParams::IngestCli => serde_json::to_value(()),
+            SourceParams::Kafka(params) => serde_json::to_value(params),
+            SourceParams::Kinesis(params) => serde_json::to_value(params),
             SourceParams::Pulsar(params) => serde_json::to_value(params),
+            SourceParams::Vec(params) => serde_json::to_value(params),
+            SourceParams::Void(params) => serde_json::to_value(params),
         }
         .unwrap()
     }
@@ -203,16 +203,16 @@ impl FromStr for SourceInputFormat {
 #[serde(tag = "source_type", content = "params", rename_all = "snake_case")]
 pub enum SourceParams {
     File(FileSourceParams),
-    Kafka(KafkaSourceParams),
     GcpPubSub(GcpPubSubSourceParams),
-    Kinesis(KinesisSourceParams),
-    Pulsar(PulsarSourceParams),
-    Vec(VecSourceParams),
-    Void(VoidSourceParams),
     #[serde(rename = "ingest-api")]
     IngestApi,
     #[serde(rename = "ingest-cli")]
     IngestCli,
+    Kafka(KafkaSourceParams),
+    Kinesis(KinesisSourceParams),
+    Pulsar(PulsarSourceParams),
+    Vec(VecSourceParams),
+    Void(VoidSourceParams),
 }
 
 impl SourceParams {
@@ -293,12 +293,12 @@ pub struct GcpPubSubSourceParams {
     #[serde(default)]
     #[serde(skip_serializing_if = "is_false")]
     pub enable_backfill_mode: bool,
-    /// GCP service account credentials (None will use default via GOOGLE_APPLICATION_CREDENTIALS)
+    /// GCP service account credentials (`None` will use default via
+    /// GOOGLE_APPLICATION_CREDENTIALS)
     pub credentials: Option<String>,
-    /// How many threads spread pubsub pull requests over (default 10)
-    /// Higher values can mean higher throughput but higher overhead
+    /// Number of pull requests issued in parallel by the source (default 1)
     pub pull_parallelism: Option<u64>,
-    /// The max messages to pull per pull request (default 1000)
+    /// Maximum number of messages returned by a pull request (default 1,000)
     pub max_messages_per_pull: Option<i32>,
 }
 

diff --git a/quickwit/quickwit-config/src/source_config/serialize.rs b/quickwit/quickwit-config/src/source_config/serialize.rs
@@ -94,18 +94,17 @@ impl SourceConfigForSerialization {
             SourceParams::Kafka(_) | SourceParams::Kinesis(_) | SourceParams::Pulsar(_) => {
                 // TODO consider any validation opportunity
             }
-            SourceParams::Vec(_)
-            | SourceParams::Void(_)
+            SourceParams::GcpPubSub(_)
             | SourceParams::IngestApi
-            | SourceParams::IngestCli => {}
-            SourceParams::GcpPubSub(_) => {}
+            | SourceParams::IngestCli
+            | SourceParams::Vec(_)
+            | SourceParams::Void(_) => {}
         }
         match &self.source_params {
-            SourceParams::Kafka(_) => {}
-            SourceParams::GcpPubSub(_) => {}
+            SourceParams::GcpPubSub(_) | SourceParams::Kafka(_) => {}
             _ => {
                 if self.desired_num_pipelines > 1 || self.max_num_pipelines_per_indexer > 1 {
-                    bail!("Quickwit currently supports multiple pipelines only for Kafka sources. Open an issue https://github.com/quickwit-oss/quickwit/issues if you need the feature for other source types.");
+                    bail!("Quickwit currently supports multiple pipelines only for GCP PubSub or Kafka sources. Open an issue https://github.com/quickwit-oss/quickwit/issues if you need the feature for other source types.");
                 }
             }
         }

diff --git a/quickwit/quickwit-indexing/Cargo.toml b/quickwit/quickwit-indexing/Cargo.toml
@@ -25,6 +25,9 @@ fail = { workspace = true }
 flume = { workspace = true }
 fnv = { workspace = true }
 futures = { workspace = true }
+google-cloud-default = { workspace = true, optional = true }
+google-cloud-googleapis = { workspace = true, optional = true }
+google-cloud-pubsub = { workspace = true, optional = true }
 itertools = { workspace = true }
 libz-sys = { workspace = true, optional = true }
 once_cell = { workspace = true }
@@ -46,9 +49,6 @@ ulid = { workspace = true }
 utoipa = { workspace = true }
 vrl = { workspace = true, optional = true }
 vrl-stdlib = { workspace = true, optional = true }
-google-cloud-pubsub = { workspace = true, optional = true }
-google-cloud-default = { workspace = true, optional = true }
-google-cloud-googleapis = { workspace = true, optional = true }
 
 quickwit-actors = { workspace = true }
 quickwit-aws = { workspace = true }
@@ -63,7 +63,7 @@ quickwit-proto = { workspace = true }
 quickwit-storage = { workspace = true }
 
 [features]
-gcp_pubsub = ["dep:google-cloud-pubsub", "dep:google-cloud-default", "dep:google-cloud-googleapis"]
+gcp-pubsub = ["dep:google-cloud-pubsub", "dep:google-cloud-default", "dep:google-cloud-googleapis"]
 kafka = ["rdkafka", "backoff"]
 kafka-broker-tests = []
 vendored-kafka = ["kafka", "libz-sys/static", "openssl/vendored", "rdkafka/gssapi-vendored"]