risingwavelabs · tabVersion · May 14, 2025 · May 14, 2025 · May 14, 2025 · May 16, 2025
diff --git a/src/connector/src/connector_common/connection.rs b/src/connector/src/connector_common/connection.rs
@@ -76,6 +76,30 @@ impl Connection for KafkaConnection {
     }
 }
 
+use rdkafka::config::RDKafkaLogLevel;
+
+pub fn read_kafka_log_level() -> RDKafkaLogLevel {
+    let log_level =
+        std::env::var("RISINGWAVE_KAFKA_LOG_LEVEL").unwrap_or_else(|_| "INFO".to_owned());
+    match log_level.to_uppercase().as_str() {
+        "DEBUG" => RDKafkaLogLevel::Debug,
+        "INFO" => RDKafkaLogLevel::Info,
+        "WARN" => RDKafkaLogLevel::Warning,
+        "ERROR" => RDKafkaLogLevel::Error,
+        "CRITICAL" => RDKafkaLogLevel::Critical,
+        "EMERG" => RDKafkaLogLevel::Emerg,
+        "ALERT" => RDKafkaLogLevel::Alert,
+        "NOTICE" => RDKafkaLogLevel::Notice,
+        _ => {
+            tracing::info!(
+                "Invalid RISINGWAVE_KAFKA_LOG_LEVEL: {}, using INFO instead",
+                log_level
+            );
+            RDKafkaLogLevel::Info
+        }
+    }
+}
+
 impl KafkaConnection {
     async fn build_client(&self) -> ConnectorResult<BaseConsumer<RwConsumerContext>> {
         let mut config = ClientConfig::new();
@@ -94,8 +118,10 @@ impl KafkaConnection {
         )
         .await?;
         let client_ctx = RwConsumerContext::new(ctx_common);
-        let client: BaseConsumer<RwConsumerContext> =
-            config.create_with_context(client_ctx).await?;
+        let client: BaseConsumer<RwConsumerContext> = config
+            .set_log_level(read_kafka_log_level())
+            .create_with_context(client_ctx)
+            .await?;
         if self.inner.is_aws_msk_iam() {
             #[cfg(not(madsim))]
             client.poll(Duration::from_secs(10)); // note: this is a blocking call

diff --git a/src/connector/src/connector_common/mod.rs b/src/connector/src/connector_common/mod.rs
@@ -25,8 +25,8 @@ pub use common::{
 };
 mod connection;
 pub use connection::{
-    validate_connection, ConfluentSchemaRegistryConnection, Connection, IcebergConnection,
-    KafkaConnection, SCHEMA_REGISTRY_CONNECTION_TYPE,
+    read_kafka_log_level, validate_connection, ConfluentSchemaRegistryConnection, Connection,
+    IcebergConnection, KafkaConnection, SCHEMA_REGISTRY_CONNECTION_TYPE,
 };
 
 mod iceberg;

diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs
@@ -35,7 +35,7 @@ use with_options::WithOptions;
 use super::catalog::{SinkFormat, SinkFormatDesc};
 use super::{Sink, SinkError, SinkParam};
 use crate::connector_common::{
-    AwsAuthProps, KafkaCommon, KafkaConnectionProps, KafkaPrivateLinkCommon,
+    read_kafka_log_level, AwsAuthProps, KafkaCommon, KafkaConnectionProps, KafkaPrivateLinkCommon,
     RdKafkaPropertiesCommon,
 };
 use crate::sink::formatter::SinkFormatterImpl;
@@ -436,7 +436,9 @@ impl KafkaSinkWriter {
             .await?;
             let producer_ctx = RwProducerContext::new(ctx_common);
             // Generate the producer
-            c.create_with_context(producer_ctx).await?
+            c.set_log_level(read_kafka_log_level())
+                .create_with_context(producer_ctx)
+                .await?
         };
 
         Ok(KafkaSinkWriter {

diff --git a/src/connector/src/source/kafka/enumerator/client.rs b/src/connector/src/source/kafka/enumerator/client.rs
@@ -27,6 +27,7 @@ use rdkafka::{ClientConfig, Offset, TopicPartitionList};
 use risingwave_common::bail;
 use risingwave_common::metrics::LabelGuardedMetric;
 
+use crate::connector_common::read_kafka_log_level;
 use crate::error::{ConnectorError, ConnectorResult};
 use crate::source::base::SplitEnumerator;
 use crate::source::kafka::split::KafkaSplit;
@@ -82,6 +83,7 @@ impl SplitEnumerator for KafkaSplitEnumerator {
         let topic = common_props.topic.clone();
         config.set("bootstrap.servers", &broker_address);
         config.set("isolation.level", KAFKA_ISOLATION_LEVEL);
+        config.set_log_level(read_kafka_log_level());
         properties.connection.set_security_properties(&mut config);
         properties.set_client(&mut config);
         let mut scan_start_offset = match properties
@@ -195,6 +197,18 @@ impl SplitEnumerator for KafkaSplitEnumerator {
             })
             .collect();
 
+        let arc_client = self.client.clone();
+        {
+            #[cfg(not(madsim))]
+            tokio::task::spawn_blocking(move || {
+                arc_client.poll(Duration::from_secs(10));
+            });
+            #[cfg(madsim)]
+            tokio::spawn(async move {
+                arc_client.poll(Duration::from_secs(10)).await;
+            });
+        }
+
         Ok(ret)
     }
 }

diff --git a/src/connector/src/source/kafka/source/reader.rs b/src/connector/src/source/kafka/source/reader.rs
@@ -22,13 +22,13 @@ use async_trait::async_trait;
 use futures::StreamExt;
 use futures_async_stream::try_stream;
 use prometheus::core::{AtomicI64, GenericGauge};
-use rdkafka::config::RDKafkaLogLevel;
 use rdkafka::consumer::{Consumer, StreamConsumer};
 use rdkafka::error::KafkaError;
 use rdkafka::{ClientConfig, Message, Offset, TopicPartitionList};
 use risingwave_common::metrics::LabelGuardedMetric;
 use risingwave_pb::plan_common::additional_column::ColumnType as AdditionalColumnType;
 
+use crate::connector_common::read_kafka_log_level;
 use crate::error::ConnectorResult as Result;
 use crate::parser::ParserConfig;
 use crate::source::base::SourceMessage;
@@ -103,10 +103,12 @@ impl SplitReader for KafkaSplitReader {
 
         let client_ctx = RwConsumerContext::new(ctx_common);
         let consumer: StreamConsumer<RwConsumerContext> = config
-            .set_log_level(RDKafkaLogLevel::Info)
+            .set_log_level(read_kafka_log_level())
             .create_with_context(client_ctx)
             .await
             .context("failed to create kafka consumer")?;
+        // poll consumer to trigger callback functions
+        let _ = tokio::time::timeout(Duration::from_millis(100), consumer.recv()).await;
 
         let mut tpl = TopicPartitionList::with_capacity(splits.len());
 

diff --git a/src/tests/simulation/tests/integration_tests/scale/sink.rs b/src/tests/simulation/tests/integration_tests/scale/sink.rs
@@ -103,6 +103,10 @@ async fn test_sink_append_only() -> Result<()> {
             }
 
             consumer.assign(&tpl).unwrap();
+
+            // poll consumer to trigger callback functions
+            let _ = tokio::time::timeout(Duration::from_millis(1000), consumer.recv()).await;
+
             consumer
         })
         .await;
@@ -185,6 +189,10 @@ async fn test_sink_debezium() -> Result<()> {
             }
 
             consumer.assign(&tpl).unwrap();
+
+            // poll consumer to trigger callback functions
+            let _ = tokio::time::timeout(Duration::from_millis(1000), consumer.recv()).await;
+
             consumer
         })
         .await;