Repository: apache/rocketmq-client-go (go)

Issue:
Message may be lost when sendMessageBack fails
The issue tracker is **ONLY** used for the go client (feature request of RocketMQ need to follow [RIP process](https://github.com/apache/rocketmq/wiki/RocketMQ-Improvement-Proposal)). Keep in mind, please check whether there is an existing same report before your raise a new one.

Alternately (especially if your communication is not a bug report), you can send mail to our [mailing lists](http://rocketmq.apache.org/about/contact/). We welcome any friendly suggestions, bug fixes, collaboration, and other improvements.

Please ensure that your bug report is clear and that it is complete. Otherwise, we may be unable to understand it or to reproduce it, either of which would prevent us from fixing the bug. We strongly recommend the report(bug report or feature request) could include some hints as to the following:

**BUG REPORT**  
1. Please describe the issue you observed:

The consumer encounters a consumption failure and returns consumer.ConsumeRetryLater, but no retry messages is received and no exception logs  are observed. The message is lost.

<img width="1329" alt="image" src="https://github.com/apache/rocketmq-client-go/assets/103550934/e9d1a53b-ebb5-4f09-9be1-61698ec7eb37">

Broker's response is not checked here and it may indicate a failure. (eg: SYSTEM_ERROR)

3. Please tell us about your environment:

     - What is your OS?
        centos

     - What is your client version?
        master branch

     - What is your RocketMQ version?
        4.x

4. Other information (e.g. detailed explanation, logs, related issues, suggestions on how to fix, etc):


Pull request description:
[ISSUE #1106] Send retry as a normal message when sendMessageBack fails
## What is the purpose of the change

fix #1106

## Brief changelog

Send retry as a normal message when sendMessageBack fails. (Refer to Java SDK)

## Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.

- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue. 
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when a cross-module dependency exists.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).


Patch:
diff --git a/consumer/consumer.go b/consumer/consumer.go
index 4ed4940..bbb5415 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -20,6 +20,7 @@ package consumer
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"sort"
 	"strconv"
 	"strings"
@@ -57,6 +58,9 @@ const (
 
 	// Offset persistent interval for consumer
 	_PersistConsumerOffsetInterval = 5 * time.Second
+
+	// Timeout for sending message to retry topic
+	_SendMessageBackAsNormalTimeout = 3 * time.Second
 )
 
 type ConsumeType string
@@ -66,6 +70,8 @@ const (
 	_PushConsume = ConsumeType("CONSUME_PASSIVELY")
 
 	_SubAll = "*"
+
+	_ClientInnerProducerGroup = "CLIENT_INNER_PRODUCER"
 )
 
 // Message model defines the way how messages are delivered to each consumer clients.
@@ -1037,6 +1043,96 @@ func (dc *defaultConsumer) searchOffsetByTimestamp(mq *primitive.MessageQueue, t
 	return strconv.ParseInt(response.ExtFields["offset"], 10, 64)
 }
 
+func (dc *defaultConsumer) sendMessageBackAsNormal(msg *primitive.MessageExt, maxReconsumeTimes int32) bool {
+	retryTopic := internal.GetRetryTopic(dc.consumerGroup)
+	normalMsg := &primitive.Message{
+		Topic: retryTopic,
+		Body:  msg.Body,
+		Flag:  msg.Flag,
+	}
+	normalMsg.WithProperties(msg.GetProperties())
+	originMsgId := msg.GetProperty(primitive.PropertyOriginMessageId)
+	if len(originMsgId) == 0 {
+		originMsgId = msg.MsgId
+	}
+	normalMsg.WithProperty(primitive.PropertyOriginMessageId, originMsgId)
+	normalMsg.WithProperty(primitive.PropertyRetryTopic, msg.Topic)
+	normalMsg.RemoveProperty(primitive.PropertyTransactionPrepared)
+	normalMsg.WithDelayTimeLevel(int(3 + msg.ReconsumeTimes))
+
+	mq, err := dc.findPublishMessageQueue(retryTopic)
+	if err != nil {
+		rlog.Warning("sendMessageBackAsNormal find publish message queue error", map[string]interface{}{
+			rlog.LogKeyTopic:         retryTopic,
+			rlog.LogKeyMessageId:     msg.MsgId,
+			rlog.LogKeyUnderlayError: err.Error(),
+		})
+		return false
+	}
+
+	brokerAddr := dc.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
+	if len(brokerAddr) == 0 {
+		rlog.Warning("sendMessageBackAsNormal cannot find broker address", map[string]interface{}{
+			rlog.LogKeyMessageId:     msg.MsgId,
+			rlog.LogKeyBroker:        mq.BrokerName,
+			rlog.LogKeyUnderlayError: err.Error(),
+		})
+		return false
+	}
+
+	request := buildSendToRetryRequest(mq, normalMsg, msg.ReconsumeTimes+1, maxReconsumeTimes)
+	resp, err := dc.client.InvokeSync(context.Background(), brokerAddr, request, _SendMessageBackAsNormalTimeout)
+	if err != nil {
+		rlog.Warning("sendMessageBackAsNormal failed to invoke", map[string]interface{}{
+			rlog.LogKeyTopic:         retryTopic,
+			rlog.LogKeyMessageId:     msg.MsgId,
+			rlog.LogKeyBroker:        brokerAddr,
+			rlog.LogKeyUnderlayError: err.Error(),
+		})
+		return false
+	}
+	if resp.Code != internal.ResSuccess {
+		rlog.Warning("sendMessageBackAsNormal failed to send", map[string]interface{}{
+			rlog.LogKeyTopic:         retryTopic,
+			rlog.LogKeyMessageId:     msg.MsgId,
+			rlog.LogKeyBroker:        brokerAddr,
+			rlog.LogKeyUnderlayError: fmt.Errorf("CODE: %d, DESC: %s", resp.Code, resp.Remark),
+		})
+		return false
+	}
+
+	return true
+}
+
+func (dc *defaultConsumer) findPublishMessageQueue(topic string) (*primitive.MessageQueue, error) {
+	mqs, err := dc.client.GetNameSrv().FetchPublishMessageQueues(topic)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(mqs) <= 0 {
+		return nil, fmt.Errorf("no writable queues")
+	}
+
+	return mqs[rand.Intn(len(mqs))], nil
+}
+
+func buildSendToRetryRequest(mq *primitive.MessageQueue, msg *primitive.Message, reconsumeTimes,
+	maxReconsumeTimes int32) *remote.RemotingCommand {
+	req := &internal.SendMessageRequestHeader{
+		ProducerGroup:     _ClientInnerProducerGroup,
+		Topic:             mq.Topic,
+		QueueId:           mq.QueueId,
+		BornTimestamp:     time.Now().UnixNano() / int64(time.Millisecond),
+		Flag:              msg.Flag,
+		Properties:        msg.MarshallProperties(),
+		ReconsumeTimes:    int(reconsumeTimes),
+		MaxReconsumeTimes: int(maxReconsumeTimes),
+	}
+
+	return remote.NewRemotingCommand(internal.ReqSendMessage, req, msg.Body)
+}
+
 func buildSubscriptionData(topic string, selector MessageSelector) *internal.SubscriptionData {
 	subData := &internal.SubscriptionData{
 		Topic:     topic,
diff --git a/consumer/option.go b/consumer/option.go
index 24acf7c..2e08163 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -382,6 +382,15 @@ func WithLimiter(limiter Limiter) Option {
 	}
 }
 
+// WithRemotingTimeout set remote client timeout options
+func WithRemotingTimeout(connectionTimeout, readTimeout, writeTimeout time.Duration) Option {
+	return func(opts *consumerOptions) {
+		opts.ClientOptions.RemotingClientConfig.ConnectionTimeout = connectionTimeout
+		opts.ClientOptions.RemotingClientConfig.ReadTimeout = readTimeout
+		opts.ClientOptions.RemotingClientConfig.WriteTimeout = writeTimeout
+	}
+}
+
 func WithTls(useTls bool) Option {
 	return func(opts *consumerOptions) {
 		opts.ClientOptions.RemotingClientConfig.UseTls = useTls
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index bd05cbf..b26c829 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -628,8 +628,12 @@ func (pc *defaultPullConsumer) sendMessageBack(brokerName string, msg *primitive
 	} else {
 		brokerAddr = msg.StoreHost
 	}
-	_, err := pc.client.InvokeSync(context.Background(), brokerAddr, pc.buildSendBackRequest(msg, delayLevel), 3*time.Second)
-	return err == nil
+	resp, err := pc.client.InvokeSync(context.Background(), brokerAddr, pc.buildSendBackRequest(msg, delayLevel), 3*time.Second)
+	if err != nil || resp.Code != internal.ResSuccess {
+		// send back as a normal message
+		return pc.defaultConsumer.sendMessageBackAsNormal(msg, pc.getMaxReconsumeTimes())
+	}
+	return true
 }
 
 func (pc *defaultPullConsumer) buildSendBackRequest(msg *primitive.MessageExt, delayLevel int) *remote.RemotingCommand {
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index ab105b3..c600f13 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -921,9 +921,10 @@ func (pc *pushConsumer) sendMessageBack(brokerName string, msg *primitive.Messag
 	} else {
 		brokerAddr = msg.StoreHost
 	}
-	_, err := pc.client.InvokeSync(context.Background(), brokerAddr, pc.buildSendBackRequest(msg, delayLevel), 3*time.Second)
-	if err != nil {
-		return false
+	resp, err := pc.client.InvokeSync(context.Background(), brokerAddr, pc.buildSendBackRequest(msg, delayLevel), 3*time.Second)
+	if err != nil || resp.Code != internal.ResSuccess {
+		// send back as a normal message
+		return pc.defaultConsumer.sendMessageBackAsNormal(msg, pc.getMaxReconsumeTimes())
 	}
 	return true
 }
diff --git a/internal/remote/remote_client.go b/internal/remote/remote_client.go
index 45dfbbf..eabfb75 100644
--- a/internal/remote/remote_client.go
+++ b/internal/remote/remote_client.go
@@ -102,7 +102,7 @@ func (c *remotingClient) InvokeSync(ctx context.Context, addr string, request *R
 	c.responseTable.Store(resp.Opaque, resp)
 	defer c.responseTable.Delete(request.Opaque)
 
-	err = c.sendRequest(conn, request)
+	err = c.sendRequest(ctx, conn, request)
 	if err != nil {
 		return nil, err
 	}
@@ -112,23 +112,24 @@ func (c *remotingClient) InvokeSync(ctx context.Context, addr string, request *R
 
 // InvokeAsync send request without blocking, just return immediately.
 func (c *remotingClient) InvokeAsync(ctx context.Context, addr string, request *RemotingCommand, callback func(*ResponseFuture)) error {
-	conn, err := c.connect(ctx, addr)
-	if err != nil {
-		return err
-	}
-
 	resp := NewResponseFuture(ctx, request.Opaque, callback)
 	c.responseTable.Store(resp.Opaque, resp)
 
-	err = c.sendRequest(conn, request)
-	if err != nil {
-		c.responseTable.Delete(request.Opaque)
-		return err
-	}
-
 	go primitive.WithRecover(func() {
+		defer resp.executeInvokeCallback()
+		defer c.responseTable.Delete(request.Opaque)
+
+		conn, err := c.connect(ctx, addr)
+		if err != nil {
+			resp.Err = err
+			return
+		}
+		err = c.sendRequest(ctx, conn, request)
+		if err != nil {
+			resp.Err = err
+			return
+		}
 		c.receiveAsync(resp)
-		c.responseTable.Delete(request.Opaque)
 	})
 
 	return nil
@@ -146,11 +147,11 @@ func (c *remotingClient) InvokeOneWay(ctx context.Context, addr string, request
 	if err != nil {
 		return err
 	}
-	return c.sendRequest(conn, request)
+	return c.sendRequest(ctx, conn, request)
 }
 
 func (c *remotingClient) connect(ctx context.Context, addr string) (*tcpConnWrapper, error) {
-	//it needs additional locker.
+	// it needs additional locker.
 	c.connectionLocker.Lock()
 	defer c.connectionLocker.Unlock()
 	conn, ok := c.connectionTable.Load(addr)
@@ -178,8 +179,21 @@ func (c *remotingClient) receiveResponse(r *tcpConnWrapper) {
 			if r.isClosed(err) {
 				return
 			}
-			if err != io.EOF {
+			// ignore name server connection read timeout
+			var isTimeout bool
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+				isTimeout = true
+			}
+			if !(err == io.EOF || isTimeout) {
 				rlog.Error("conn error, close connection", map[string]interface{}{
+					"remoteAddr":             r.RemoteAddr(),
+					"localAddr":              r.LocalAddr(),
+					rlog.LogKeyUnderlayError: err,
+				})
+			} else {
+				rlog.Debug("conn error, close connection", map[string]interface{}{
+					"remoteAddr":             r.RemoteAddr(),
+					"localAddr":              r.LocalAddr(),
 					rlog.LogKeyUnderlayError: err,
 				})
 			}
@@ -230,10 +244,10 @@ func (c *remotingClient) processCMD(cmd *RemotingCommand, r *tcpConnWrapper) {
 			responseFuture := resp.(*ResponseFuture)
 			go primitive.WithRecover(func() {
 				responseFuture.ResponseCommand = cmd
-				responseFuture.executeInvokeCallback()
 				if responseFuture.Done != nil {
 					close(responseFuture.Done)
 				}
+				responseFuture.executeInvokeCallback()
 			})
 		}
 	} else {
@@ -246,7 +260,7 @@ func (c *remotingClient) processCMD(cmd *RemotingCommand, r *tcpConnWrapper) {
 				if res != nil {
 					res.Opaque = cmd.Opaque
 					res.Flag |= 1 << 0
-					err := c.sendRequest(r, res)
+					err := c.sendRequest(context.Background(), r, res)
 					if err != nil {
 						rlog.Warning("send response to broker error", map[string]interface{}{
 							rlog.LogKeyUnderlayError: err,
@@ -297,23 +311,27 @@ func (c *remotingClient) createScanner(r io.Reader) *bufio.Scanner {
 	return scanner
 }
 
-func (c *remotingClient) sendRequest(conn *tcpConnWrapper, request *RemotingCommand) error {
+func (c *remotingClient) sendRequest(ctx context.Context, conn *tcpConnWrapper, request *RemotingCommand) error {
 	var err error
 	if c.interceptor != nil {
-		err = c.interceptor(context.Background(), request, nil, func(ctx context.Context, req, reply interface{}) error {
-			return c.doRequest(conn, request)
+		err = c.interceptor(ctx, request, nil, func(ctx context.Context, req, reply interface{}) error {
+			return c.doRequest(ctx, conn, request)
 		})
 	} else {
-		err = c.doRequest(conn, request)
+		err = c.doRequest(ctx, conn, request)
 	}
 	return err
 }
 
-func (c *remotingClient) doRequest(conn *tcpConnWrapper, request *RemotingCommand) error {
+func (c *remotingClient) doRequest(ctx context.Context, conn *tcpConnWrapper, request *RemotingCommand) error {
 	conn.Lock()
 	defer conn.Unlock()
 
-	err := conn.Conn.SetWriteDeadline(time.Now().Add(c.config.WriteTimeout))
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		deadline = time.Now().Add(c.config.WriteTimeout)
+	}
+	err := conn.Conn.SetWriteDeadline(deadline)
 	if err != nil {
 		rlog.Error("conn error, close connection", map[string]interface{}{
 			rlog.LogKeyUnderlayError: err,
diff --git a/primitive/result.go b/primitive/result.go
index a20a448..5147ebd 100644
--- a/primitive/result.go
+++ b/primitive/result.go
@@ -58,8 +58,12 @@ func NewSendResult() *SendResult {
 
 // SendResult send message result to string(detail result)
 func (result *SendResult) String() string {
+	mq := "nil"
+	if result.MessageQueue != nil {
+		mq = result.MessageQueue.String()
+	}
 	return fmt.Sprintf("SendResult [sendStatus=%d, msgIds=%s, offsetMsgId=%s, queueOffset=%d, messageQueue=%s]",
-		result.Status, result.MsgID, result.OffsetMsgID, result.QueueOffset, result.MessageQueue.String())
+		result.Status, result.MsgID, result.OffsetMsgID, result.QueueOffset, mq)
 }
 
 // TransactionSendResult RocketMQ send result
diff --git a/producer/option.go b/producer/option.go
index 6e43cc2..72af3c6 100644
--- a/producer/option.go
+++ b/producer/option.go
@@ -179,6 +179,15 @@ func WithCompressLevel(level int) Option {
 	}
 }
 
+// WithRemotingTimeout set remote client timeout options
+func WithRemotingTimeout(connectionTimeout, readTimeout, writeTimeout time.Duration) Option {
+	return func(opts *producerOptions) {
+		opts.ClientOptions.RemotingClientConfig.ConnectionTimeout = connectionTimeout
+		opts.ClientOptions.RemotingClientConfig.ReadTimeout = readTimeout
+		opts.ClientOptions.RemotingClientConfig.WriteTimeout = writeTimeout
+	}
+}
+
 func WithTls(useTls bool) Option {
 	return func(opts *producerOptions) {
 		opts.ClientOptions.RemotingClientConfig.UseTls = useTls
diff --git a/producer/producer.go b/producer/producer.go
index f823884..eb3cd2e 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -26,14 +26,15 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+
 	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/internal/utils"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/rlog"
-	"github.com/google/uuid"
-	"github.com/pkg/errors"
 )
 
 type defaultProducer struct {
@@ -328,7 +329,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 		if mq != nil {
 			lastBrokerName = mq.BrokerName
 		}
-		mq := p.selectMessageQueue(msg, lastBrokerName)
+		mq = p.selectMessageQueue(msg, lastBrokerName)
 		if mq == nil {
 			err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
 			continue
@@ -355,7 +356,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 			producerCtx.MQ = *mq
 		}
 
-		res, _err := p.client.InvokeSync(ctx, addr, p.buildSendRequest(mq, msg), 3*time.Second)
+		res, _err := p.client.InvokeSync(ctx, addr, p.buildSendRequest(mq, msg), p.options.SendMsgTimeout)
 		if _err != nil {
 			err = _err
 			continue
@@ -400,7 +401,7 @@ func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message,
 		return errors.Errorf("topic=%s route info not found", mq.Topic)
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	ctx, cancel := context.WithTimeout(ctx, p.options.SendMsgTimeout)
 	err := p.client.InvokeAsync(ctx, addr, p.buildSendRequest(mq, msg), func(command *remote.RemotingCommand, err error) {
 		cancel()
 		if err != nil {
@@ -465,7 +466,7 @@ func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
 
-		_err := p.client.InvokeOneWay(ctx, addr, p.buildSendRequest(mq, msg), 3*time.Second)
+		_err := p.client.InvokeOneWay(ctx, addr, p.buildSendRequest(mq, msg), p.options.SendMsgTimeout)
 		if _err != nil {
 			err = _err
 			continue


Test patch:
diff --git a/consumer/option_test.go b/consumer/option_test.go
index ab99b63..4db5b93 100644
--- a/consumer/option_test.go
+++ b/consumer/option_test.go
@@ -3,6 +3,7 @@ package consumer
 import (
 	"reflect"
 	"testing"
+	"time"
 )
 
 func getFieldString(obj interface{}, field string) string {
@@ -12,6 +13,20 @@ func getFieldString(obj interface{}, field string) string {
 	}).String()
 }
 
+func TestWithRemotingTimeout(t *testing.T) {
+	opt := defaultPushConsumerOptions()
+	WithRemotingTimeout(3*time.Second, 4*time.Second, 5*time.Second)(&opt)
+	if timeout := opt.RemotingClientConfig.ConnectionTimeout; timeout != 3*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout connectionTimeout. want:%s, got=%s", 3*time.Second, timeout)
+	}
+	if timeout := opt.RemotingClientConfig.ReadTimeout; timeout != 4*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout readTimeout. want:%s, got=%s", 4*time.Second, timeout)
+	}
+	if timeout := opt.RemotingClientConfig.WriteTimeout; timeout != 5*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout writeTimeout. want:%s, got=%s", 5*time.Second, timeout)
+	}
+}
+
 func TestWithUnitName(t *testing.T) {
 	opt := defaultPushConsumerOptions()
 	unitName := "unsh"
diff --git a/producer/option_test.go b/producer/option_test.go
index 723da03..9b6ee13 100644
--- a/producer/option_test.go
+++ b/producer/option_test.go
@@ -3,6 +3,7 @@ package producer
 import (
 	"reflect"
 	"testing"
+	"time"
 )
 
 func getFieldString(obj interface{}, field string) string {
@@ -12,6 +13,20 @@ func getFieldString(obj interface{}, field string) string {
 	}).String()
 }
 
+func TestWithRemotingTimeout(t *testing.T) {
+	opt := defaultProducerOptions()
+	WithRemotingTimeout(3*time.Second, 4*time.Second, 5*time.Second)(&opt)
+	if timeout := opt.RemotingClientConfig.ConnectionTimeout; timeout != 3*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout connectionTimeout. want:%s, got=%s", 3*time.Second, timeout)
+	}
+	if timeout := opt.RemotingClientConfig.ReadTimeout; timeout != 4*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout readTimeout. want:%s, got=%s", 4*time.Second, timeout)
+	}
+	if timeout := opt.RemotingClientConfig.WriteTimeout; timeout != 5*time.Second {
+		t.Errorf("consumer option WithRemotingTimeout writeTimeout. want:%s, got=%s", 5*time.Second, timeout)
+	}
+}
+
 func TestWithUnitName(t *testing.T) {
 	opt := defaultProducerOptions()
 	unitName := "unsh"