From d84cda22adc4e233f9a6ae7553d50da6dacfb7a2 Mon Sep 17 00:00:00 2001
From: jaime <yun.zhang@zilliz.com>
Date: Mon, 21 Nov 2022 17:49:12 +0800
Subject: [PATCH] Fix start fail with kafka model (#20726)

Signed-off-by: yun.zhang <yun.zhang@zilliz.com>

Signed-off-by: yun.zhang <yun.zhang@zilliz.com>
---
 internal/util/paramtable/service_param.go     |  4 +++-
 .../util/paramtable/service_param_test.go     | 21 +++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/internal/util/paramtable/service_param.go b/internal/util/paramtable/service_param.go
index b3604a3207..86968f2b89 100644
--- a/internal/util/paramtable/service_param.go
+++ b/internal/util/paramtable/service_param.go
@@ -351,10 +351,11 @@ func (p *PulsarConfig) Init(base *BaseTable) {
 	}
 	p.Port.Init(base.mgr)
 
+	// due to implicit rule of MQ priority，the default address should be empty
 	p.Address = ParamItem{
 		Key:          "pulsar.address",
 		Version:      "2.0.0",
-		DefaultValue: "localhost",
+		DefaultValue: "",
 		Formatter: func(addr string) string {
 			if addr == "" {
 				return ""
@@ -453,6 +454,7 @@ type KafkaConfig struct {
 }
 
 func (k *KafkaConfig) Init(base *BaseTable) {
+	// due to implicit rule of MQ priority，the default address should be empty
 	k.Address = ParamItem{
 		Key:          "kafka.brokerList",
 		DefaultValue: "",
diff --git a/internal/util/paramtable/service_param_test.go b/internal/util/paramtable/service_param_test.go
index 26c159ad34..98e189f428 100644
--- a/internal/util/paramtable/service_param_test.go
+++ b/internal/util/paramtable/service_param_test.go
@@ -14,6 +14,8 @@ package paramtable
 import (
 	"testing"
 
+	"github.com/milvus-io/milvus/internal/config"
+
 	"github.com/milvus-io/milvus/internal/util/metricsinfo"
 	"github.com/stretchr/testify/assert"
 )
@@ -60,6 +62,13 @@ func TestServiceParam(t *testing.T) {
 	})
 
 	t.Run("test pulsarConfig", func(t *testing.T) {
+		// test default value
+		{
+			pc := &PulsarConfig{}
+			base := &BaseTable{mgr: &config.Manager{}}
+			pc.Init(base)
+			assert.Empty(t, pc.Address.GetValue())
+		}
 		{
 			assert.NotEqual(t, SParams.PulsarCfg.Address.GetValue(), "")
 			t.Logf("pulsar address = %s", SParams.PulsarCfg.Address.GetValue())
@@ -118,6 +127,18 @@ func TestServiceParam(t *testing.T) {
 		t.Logf("rocksmq path = %s", Params.Path.GetValue())
 	})
 
+	t.Run("test kafkaConfig", func(t *testing.T) {
+		// test default value
+		{
+			kc := &KafkaConfig{}
+			base := &BaseTable{mgr: &config.Manager{}}
+			kc.Init(base)
+			assert.Empty(t, kc.Address.GetValue())
+			assert.Equal(t, kc.SaslMechanisms.GetValue(), "PLAIN")
+			assert.Equal(t, kc.SecurityProtocol.GetValue(), "SASL_SSL")
+		}
+	})
+
 	t.Run("test minioConfig", func(t *testing.T) {
 		Params := &SParams.MinioCfg