diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index a8307e7de1..7e0ee4e536 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -770,6 +770,7 @@ DEFINE_mInt32(max_remote_storage_count, "10");
 // and the valid values are: 0.9.0.x, 0.8.x.y.
 DEFINE_String(kafka_api_version_request, "true");
 DEFINE_String(kafka_broker_version_fallback, "0.10.0");
+DEFINE_String(kafka_debug, "disable");
 
 // The number of pool siz of routine load consumer.
 // If you meet the error describe in https://github.com/edenhill/librdkafka/issues/3608
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 8d978c768e..89aa6d4e14 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -826,6 +826,7 @@ DECLARE_mInt32(max_remote_storage_count);
 // and the valid values are: 0.9.0.x, 0.8.x.y.
 DECLARE_String(kafka_api_version_request);
 DECLARE_String(kafka_broker_version_fallback);
+DECLARE_mString(kafka_debug);
 
 // The number of pool siz of routine load consumer.
 // If you meet the error describe in https://github.com/edenhill/librdkafka/issues/3608
diff --git a/be/src/runtime/routine_load/data_consumer.cpp b/be/src/runtime/routine_load/data_consumer.cpp
index cdbff1c1d1..3706e31fb4 100644
--- a/be/src/runtime/routine_load/data_consumer.cpp
+++ b/be/src/runtime/routine_load/data_consumer.cpp
@@ -91,6 +91,10 @@ Status KafkaDataConsumer::init(std::shared_ptr<StreamLoadContext> ctx) {
     RETURN_IF_ERROR(set_conf("api.version.request", config::kafka_api_version_request));
     RETURN_IF_ERROR(set_conf("api.version.fallback.ms", "0"));
     RETURN_IF_ERROR(set_conf("broker.version.fallback", config::kafka_broker_version_fallback));
+    RETURN_IF_ERROR(set_conf("broker.address.ttl", "0"));
+    if (config::kafka_debug != "disable") {
+        RETURN_IF_ERROR(set_conf("debug", config::kafka_debug));
+    }
 
     for (auto& item : ctx->kafka_info->properties) {
         if (starts_with(item.second, "FILE:")) {