You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2021/07/22 06:44:00 UTC
[GitHub] [incubator-inlong] charlely commented on a change in pull request #573: [INLONG-778]Fix Go SDK Consumer Bug
charlely commented on a change in pull request #573:
URL: https://github.com/apache/incubator-inlong/pull/573#discussion_r674529777
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -120,7 +120,8 @@ func (c *consumer) register2Master(needChange bool) error {
}
c.master = node
}
- for c.master.HasNext {
+ retryCount := 0
+ for {
rsp, err := c.sendRegRequest2Master()
if err != nil {
Review comment:
Why here if get error will return err, but not use next master address to register?
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -223,17 +228,18 @@ func (c *consumer) GetMessage() (*ConsumerResult, error) {
rsp, err := c.client.GetMessageRequestC2B(ctx, m, c.subInfo, c.rmtDataCache)
if err != nil {
log.Infof("[CONSUMER]GetMessage error %s", err.Error())
+ err1 := c.rmtDataCache.ReleasePartition(true, isFiltered, confirmContext, false)
Review comment:
Why to ReleasePartition?
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -223,17 +228,18 @@ func (c *consumer) GetMessage() (*ConsumerResult, error) {
rsp, err := c.client.GetMessageRequestC2B(ctx, m, c.subInfo, c.rmtDataCache)
if err != nil {
log.Infof("[CONSUMER]GetMessage error %s", err.Error())
+ err1 := c.rmtDataCache.ReleasePartition(true, isFiltered, confirmContext, false)
Review comment:
This way can use like it:
if err := c.rmtDataCache.ReleasePartition(true, isFiltered, confirmContext, false); err != nil {
}
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -619,34 +625,43 @@ func (c *consumer) convertMessages(filtered bool, topic string, rsp *protocol.Ge
msgs := make([]*Message, 0, len(rsp.GetMessages()))
for _, m := range rsp.GetMessages() {
- checkSum := uint64(crc32.Update(0, crc32.IEEETable, m.GetPayLoadData())) & 0x7FFFFFFFF
+ checkSum := uint64(crc32.Update(0, crc32.IEEETable, m.GetPayLoadData())) & 0x7FFFFFFF
if int32(checkSum) != m.GetCheckSum() {
continue
}
readPos := 0
- dataLen := len(m.GetPayLoadData())
+ payLoadData := m.GetPayLoadData()
+ dataLen := len(payLoadData)
var properties map[string]string
if m.GetFlag()&0x01 == 1 {
Review comment:
Cann't use magic number to here.
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -120,7 +120,8 @@ func (c *consumer) register2Master(needChange bool) error {
}
c.master = node
}
- for c.master.HasNext {
+ retryCount := 0
+ for {
rsp, err := c.sendRegRequest2Master()
if err != nil {
log.Infof("[CONSUMER]register2Master error %s", err.Error())
Review comment:
Here need use Error level log.
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -120,7 +120,8 @@ func (c *consumer) register2Master(needChange bool) error {
}
c.master = node
}
- for c.master.HasNext {
+ retryCount := 0
+ for {
rsp, err := c.sendRegRequest2Master()
if err != nil {
Review comment:
If first address net error, will always return error ?
##########
File path: tubemq-client-twins/tubemq-client-go/client/consumer_impl.go
##########
@@ -143,7 +148,7 @@ func (c *consumer) register2Master(needChange bool) error {
c.masterHBRetry = 0
c.processRegisterResponseM2C(rsp)
- return nil
+ break
}
Review comment:
If last fail to register to master, need printf fail result to Error log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org