diff --git a/src/common/wal/src/config.rs b/src/common/wal/src/config.rs
index 017eb56e33..5a0d5db426 100644
--- a/src/common/wal/src/config.rs
+++ b/src/common/wal/src/config.rs
@@ -268,4 +268,26 @@ mod tests {
         };
         assert_eq!(datanode_wal_config, DatanodeWalConfig::Kafka(expected));
     }
+
+    #[test]
+    fn test_kafka_wal_config_debug_redacts_password() {
+        let config = MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
+            connection: KafkaConnectionConfig {
+                sasl: Some(KafkaClientSasl {
+                    config: KafkaClientSaslConfig::Plain {
+                        username: "greptime".to_string(),
+                        password: "kafka-secret".to_string(),
+                    },
+                }),
+                ..Default::default()
+            },
+            ..Default::default()
+        });
+
+        let debug = format!("{config:#?}");
+
+        assert!(debug.contains("greptime"));
+        assert!(debug.contains("<REDACTED>"));
+        assert!(!debug.contains("kafka-secret"));
+    }
 }
diff --git a/src/common/wal/src/config/kafka/common.rs b/src/common/wal/src/config/kafka/common.rs
index 8840197086..4266f30a85 100644
--- a/src/common/wal/src/config/kafka/common.rs
+++ b/src/common/wal/src/config/kafka/common.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::io::Cursor;
 use std::sync::Arc;
 use std::time::Duration;
@@ -63,7 +64,7 @@ pub struct KafkaClientSasl {
     pub config: KafkaClientSaslConfig,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(tag = "type", rename_all = "SCREAMING-KEBAB-CASE")]
 pub enum KafkaClientSaslConfig {
     Plain {
@@ -82,6 +83,28 @@ pub enum KafkaClientSaslConfig {
     },
 }
 
+impl fmt::Debug for KafkaClientSaslConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            KafkaClientSaslConfig::Plain { username, .. } => f
+                .debug_struct("Plain")
+                .field("username", username)
+                .field("password", &"<REDACTED>")
+                .finish(),
+            KafkaClientSaslConfig::ScramSha256 { username, .. } => f
+                .debug_struct("ScramSha256")
+                .field("username", username)
+                .field("password", &"<REDACTED>")
+                .finish(),
+            KafkaClientSaslConfig::ScramSha512 { username, .. } => f
+                .debug_struct("ScramSha512")
+                .field("username", username)
+                .field("password", &"<REDACTED>")
+                .finish(),
+        }
+    }
+}
+
 impl KafkaClientSaslConfig {
     /// Converts to [`SaslConfig`].
     pub fn into_sasl_config(self) -> SaslConfig {