[INLONG-1525]Fix bug which will cause Go SDK fail to parse SubscribeI…

…nfo (#1526)

Signed-off-by: Zijie Lu <wslzj40@gmail.com>

tubemq-client-twins/tubemq-client-go/client/consumer_impl.go

@@ -162,6 +162,10 @@ func (c *consumer) sendRegRequest2Master() (*protocol.RegisterResponseM2C, error
 	node := &metadata.Node{}
 	node.SetHost(util.GetLocalHost())
 	node.SetAddress(c.master.Address)
+	auth := &protocol.AuthenticateInfo{}
+	if c.needGenMasterCertificateInfo(true) {
+		util.GenMasterAuthenticateToken(auth, c.config.Net.Auth.UserName, c.config.Net.Auth.Password)
+	}
 	m.SetNode(node)
 	sub := &metadata.SubscribeInfo{}
 	sub.SetGroup(c.config.Consumer.Group)
@@ -512,7 +516,7 @@ func (c *consumer) genBrokerAuthenticInfo(force bool) *protocol.AuthorizedInfo {
 	return auth
 }
 
-func (c *consumer) genMasterAuthenticateToken(auth *protocol.AuthenticateInfo, force bool) {
+func (c *consumer) needGenMasterCertificateInfo(force bool) bool {
 	needAdd := false
 	if c.config.Net.Auth.Enable {
 		if force {
@@ -526,6 +530,7 @@ func (c *consumer) genMasterAuthenticateToken(auth *protocol.AuthenticateInfo, f
 		if needAdd {
 		}
 	}
+	return needAdd
 }
 
 func (c *consumer) getConsumeReadStatus(isFirstReg bool) int32 {
@@ -700,10 +705,10 @@ func (c *consumer) close2Master() error {
 	sub := &metadata.SubscribeInfo{}
 	sub.SetGroup(c.config.Consumer.Group)
 	m.SetSubscribeInfo(sub)
+	mci := &protocol.MasterCertificateInfo{}
 	auth := &protocol.AuthenticateInfo{}
-	c.genMasterAuthenticateToken(auth, true)
-	mci := &protocol.MasterCertificateInfo{
-		AuthInfo: auth,
+	if c.needGenMasterCertificateInfo(true) {
+		util.GenMasterAuthenticateToken(auth, c.config.Net.Auth.UserName, c.config.Net.Auth.Password)
 	}
 	c.subInfo.SetMasterCertificateInfo(mci)
 	rsp, err := c.client.CloseRequestC2M(ctx, m, c.subInfo)

tubemq-client-twins/tubemq-client-go/client/heartbeat.go

@@ -88,6 +88,10 @@ func (h *heartbeatManager) consumerHB2Master() {
 	sub := &metadata.SubscribeInfo{}
 	sub.SetGroup(h.consumer.config.Consumer.Group)
 	m.SetSubscribeInfo(sub)
+	auth := &protocol.AuthenticateInfo{}
+	if h.consumer.needGenMasterCertificateInfo(true) {
+		util.GenMasterAuthenticateToken(auth, h.consumer.config.Net.Auth.UserName, h.consumer.config.Net.Auth.Password)
+	}
 	h.consumer.unreportedTimes++
 	if h.consumer.unreportedTimes > h.consumer.config.Consumer.MaxSubInfoReportInterval {
 		m.SetReportTimes(true)

tubemq-client-twins/tubemq-client-go/metadata/subscribe_info.go

@@ -55,7 +55,7 @@ func (s *SubscribeInfo) String() string {
 // If the given is invalid, it will return error.
 // The format of subscribeInfo string: consumerId@group#broker_info#topic:partitionId
 func NewSubscribeInfo(subscribeInfo string) (*SubscribeInfo, error) {
-	s := strings.Split(subscribeInfo, "#")
+	s := strings.SplitN(subscribeInfo, "#", 2)
 	if len(s) == 1 {
 		return nil, errs.ErrInvalidSubscribeInfoString
 	}

tubemq-client-twins/tubemq-client-go/remote/remote.go

@@ -463,7 +463,7 @@ func (r *RmtDataCache) removeFromIndexPartitions(partitionKey string) {
 		r.indexPartitions = nil
 		return
 	}
-	if pos >= len(r.indexPartitions) {
+	if pos == -1 || pos >= len(r.indexPartitions) {
 		return
 	}
 	r.indexPartitions = append(r.indexPartitions[:pos], r.indexPartitions[pos+1:]...)

tubemq-client-twins/tubemq-client-go/util/util.go

@@ -23,6 +23,8 @@ import (
 	"net"
 	"strconv"
 	"strings"
+
+	"github.com/apache/incubator-inlong/tubemq-client-twins/tubemq-client-go/protocol"
 )
 
 // InvalidValue defines the invalid value of TubeMQ config.
@@ -72,8 +74,7 @@ func GenBrokerAuthenticateToken(username string, password string) string {
 }
 
 // GenMasterAuthenticateToken generates the master authenticate token.
-func GenMasterAuthenticateToken(username string, password string) string {
-	return ""
+func GenMasterAuthenticateToken(authInfo *protocol.AuthenticateInfo, username string, password string) {
 }
 
 // ParseConfirmContext parses the confirm context to partition key and bookedTime.