Fix Kafka tests

Author: guilload

quickwit-indexing/Cargo.toml

@@ -45,10 +45,10 @@ arc-swap = "1.4"
 
 [features]
 kafka = ["rdkafka"]
-kafka-broker-external-service = []
+kafka-broker-tests = []
 vendored-kafka = ["kafka", "libz-sys/static", "openssl/vendored"]
 kinesis = ["rusoto_core", "rusoto_kinesis"]
-kinesis-external-service = []
+kinesis-localstack-tests = []
 
 [dev-dependencies]
 bytes = "1"

quickwit-indexing/src/source/kafka_source.rs

@@ -656,10 +656,11 @@ mod tests {
     }
 }
 
-#[cfg(all(test, feature = "kafka-external-service"))]
-mod kafka_source_tests {
+#[cfg(all(test, feature = "kafka-broker-tests"))]
+mod kafka_broker_tests {
     use quickwit_actors::{create_test_mailbox, Universe};
     use quickwit_common::rand::append_random_suffix;
+    use quickwit_config::SourceParams;
     use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
     use rdkafka::client::DefaultClientContext;
     use rdkafka::message::ToBytes;
@@ -802,18 +803,18 @@ mod kafka_source_tests {
         create_topic(&admin_client, &topic, 3).await?;
 
         let source_config = SourceConfig {
-            source_id: "kafka-test-source".to_string(),
-            source_type: "kafka".to_string(),
-            params: json!({
-                "topic": topic,
-                "client_log_level": "info",
-                "client_params": json!({
+            source_id: "test-kafka-source".to_string(),
+            source_params: SourceParams::Kafka(KafkaSourceParams {
+                topic: topic.clone(),
+                client_log_level: None,
+                client_params: json!({
                     "bootstrap.servers": bootstrap_servers,
                     "group.id": group_id,
                     "enable.partition.eof": true,
                 }),
             }),
         };
+
         let source_loader = quickwit_supported_sources();
         {
             let (sink, inbox) = create_test_mailbox();

quickwit-indexing/src/source/kinesis/api.rs

@@ -265,7 +265,7 @@ pub(crate) mod tests {
     }
 }
 
-#[cfg(all(test, feature = "kinesis-external-service"))]
+#[cfg(all(test, feature = "kinesis-localstack-tests"))]
 mod kinesis_localstack_tests {
     use std::time::Duration;
 

quickwit-indexing/src/source/kinesis/shard_consumer.rs

@@ -189,7 +189,7 @@ impl AsyncActor for ShardConsumer {
     }
 }
 
-#[cfg(all(test, feature = "kinesis-external-service"))]
+#[cfg(all(test, feature = "kinesis-localstack-tests"))]
 mod kinesis_localstack_tests {
     use quickwit_actors::{create_test_mailbox, Universe};