diff --git a/configs/milvus.yaml b/configs/milvus.yaml
index e079cf1095..665c617c32 100644
--- a/configs/milvus.yaml
+++ b/configs/milvus.yaml
@@ -123,8 +123,8 @@ pulsar:
 #   brokerList:
 #   saslUsername:
 #   saslPassword:
-#   saslMechanisms: PLAIN
-#   securityProtocol: SASL_SSL
+#   saslMechanisms:
+#   securityProtocol:
 #   readTimeout: 10 # read message timeout in seconds
 #   ssl:
 #     enabled: false # Whether to support kafka secure connection mode
diff --git a/pkg/util/paramtable/service_param.go b/pkg/util/paramtable/service_param.go
index 94b08e4807..fee6b55e0c 100644
--- a/pkg/util/paramtable/service_param.go
+++ b/pkg/util/paramtable/service_param.go
@@ -722,7 +722,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.SaslMechanisms = ParamItem{
 		Key:          "kafka.saslMechanisms",
-		DefaultValue: "PLAIN",
+		DefaultValue: "",
 		Version:      "2.1.0",
 		Export:       true,
 	}
@@ -730,7 +730,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.SecurityProtocol = ParamItem{
 		Key:          "kafka.securityProtocol",
-		DefaultValue: "SASL_SSL",
+		DefaultValue: "",
 		Version:      "2.1.0",
 		Export:       true,
 	}
@@ -739,7 +739,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 	k.KafkaUseSSL = ParamItem{
 		Key:          "kafka.ssl.enabled",
 		DefaultValue: "false",
-		Version:      "2.3.8",
+		Version:      "2.3.11",
 		Doc:          "whether to enable ssl mode",
 		Export:       true,
 	}
@@ -747,7 +747,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.KafkaTLSCert = ParamItem{
 		Key:     "kafka.ssl.tlsCert",
-		Version: "2.3.8",
+		Version: "2.3.11",
 		Doc:     "path to client's public key (PEM) used for authentication",
 		Export:  true,
 	}
@@ -755,7 +755,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.KafkaTLSKey = ParamItem{
 		Key:     "kafka.ssl.tlsKey",
-		Version: "2.3.8",
+		Version: "2.3.11",
 		Doc:     "path to client's private key (PEM) used for authentication",
 		Export:  true,
 	}
@@ -763,7 +763,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.KafkaTLSCACert = ParamItem{
 		Key:     "kafka.ssl.tlsCaCert",
-		Version: "2.3.8",
+		Version: "2.3.11",
 		Doc:     "file or directory path to CA certificate(s) for verifying the broker's key",
 		Export:  true,
 	}
@@ -771,7 +771,7 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 
 	k.KafkaTLSKeyPassword = ParamItem{
 		Key:     "kafka.ssl.tlsKeyPassword",
-		Version: "2.3.8",
+		Version: "2.3.11",
 		Doc:     "private key passphrase for use with ssl.key.location and set_ssl_cert(), if any",
 		Export:  true,
 	}
diff --git a/pkg/util/paramtable/service_param_test.go b/pkg/util/paramtable/service_param_test.go
index 6c56484b98..ae00979fbf 100644
--- a/pkg/util/paramtable/service_param_test.go
+++ b/pkg/util/paramtable/service_param_test.go
@@ -166,8 +166,8 @@ func TestServiceParam(t *testing.T) {
 			base := &BaseTable{mgr: config.NewManager()}
 			kc.Init(base)
 			assert.Empty(t, kc.Address.GetValue())
-			assert.Equal(t, kc.SaslMechanisms.GetValue(), "PLAIN")
-			assert.Equal(t, kc.SecurityProtocol.GetValue(), "SASL_SSL")
+			assert.Empty(t, kc.SaslMechanisms.GetValue())
+			assert.Empty(t, kc.SecurityProtocol.GetValue())
 			assert.Equal(t, kc.ReadTimeout.GetAsDuration(time.Second), 10*time.Second)
 			assert.Equal(t, kc.KafkaUseSSL.GetAsBool(), false)
 			assert.Empty(t, kc.KafkaTLSCACert.GetValue())