This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 3c0849e9 fix(go): robust consumer settlement, reception backoff, and
shutdown under failures (#1473)
3c0849e9 is described below
commit 3c0849e992771986b5690f6a169d4afdcef0b2f7
Author: zhaohai <[email protected]>
AuthorDate: Mon Nov 2 17:38:58 2026 +0800
fix(go): robust consumer settlement, reception backoff, and shutdown under
failures (#1473)
* fix(go): give up dead settlement retries and stop swallowing settlement
errors
- move the INVALID_RECEIPT_HANDLE abandon branch of
defaultProcessQueue.changeInvisibleDuration before the generic err
handling: changeInvisibleDuration0 converts a non-OK status into an
error, so the branch was unreachable and an expired receipt handle
was renewed once a second forever
- cap the ack / change invisible duration / forward-to-DLQ retry chains
at 3 attempts and stop scheduling retries once the process queue was
dropped or the consumer is no longer running; the DLQ chain also
forgives INVALID_RECEIPT_HANDLE (it never gave up on any code)
- restore the Java semantics of the receiving back-off delays: the
default is RECEIVING_FAILURE_BACKOFF_DELAY (1s) and only flow control
retries after the short RECEIVING_FLOW_CONTROL_BACKOFF_DELAY (20ms)
- SimpleConsumer.ChangeInvisibleDuration: only adopt the receipt handle
from a successful response; a failed request used to wipe the handle
to the empty string and make every following call fail permanently
- SimpleConsumer.Ack: return the server status instead of nil on a
non-OK response
- fix mis-copied "message acknowledgement" log text in the renewal and
DLQ failure paths and replace %w verbs that zap does not expand
* fix(go): stop receive hammering, throttle reception failure logs, and
bound GracefulStop under proxy restart
- Remove the 20ms TOO_MANY_REQUESTS reception backoff: real proxy flow
control means the executor queue is saturated, and retrying at ~50
polls per second only hammers it harder; every reception failure now
waits the full 1s backoff.
- Throttle reception-failure logging: the first failure logs ERROR, then
one WARN per 30 consecutive failures carrying the cumulative count,
the rest only DEBUG; the counter resets on success and on
MESSAGE_NOT_FOUND, so a sustained outage no longer emits one ERROR per
second per queue.
- GracefulStop now cancels in-flight long polling receive requests via
a consumer-scoped receive context, so shutdown no longer waits out
requestTimeout + longPollingTimeout for a poll hanging against a
restarting proxy.
- notifyClientTermination runs concurrently across route targets, so the
shutdown no longer stretches by a full request timeout per target.
- Replace remaining Java-style {} placeholders with fmt verbs in
process_queue.go and push_consumer.go logs.
- Add tests: ack settlement retries are capped and complete; GracefulStop
cancels an in-flight receive request instead of waiting it out.
* fix(go): keep assignments when the assignment scan fails and fix clientId
verb
A failed queryAssignments does not mean the server has no assignments:
syncing the nil result dropped every working process queue and stalled
consumption for the whole outage. The scan now keeps the current
assignments until the next successful one, so the reception chain keeps
running (and its failure logging stays throttled) while the proxy or
broker is down.
Also fix clientId=%d printing %!d(string=...) in the empty-assignments
warning, found during the local cluster test.
---
golang/client.go | 19 ++-
golang/process_queue.go | 215 ++++++++++++++++++++++++----------
golang/process_queue_test.go | 260 +++++++++++++++++++++++++++++++++++++++++
golang/push_consumer.go | 26 ++++-
golang/push_consumer_test.go | 21 ++++
golang/simple_consumer.go | 12 +-
golang/simple_consumer_test.go | 89 ++++++++++++++
7 files changed, 568 insertions(+), 74 deletions(-)
diff --git a/golang/client.go b/golang/client.go
index 4a83d29a..6dc5e22c 100644
--- a/golang/client.go
+++ b/golang/client.go
@@ -802,17 +802,26 @@ func (cli *defaultClient) notifyClientTermination() {
},
}
targets := cli.getTotalTargets()
+ // The notifications run concurrently: against a restarting proxy every
+ // target needs its full request timeout, and issuing them sequentially
+ // would stretch the shutdown by that timeout per route target.
+ var wg sync.WaitGroup
for _, target := range targets {
endpoints, _ := utils.ParseTarget(target)
if endpoints == nil {
continue
}
- cli.log.Infof("start notifyClientTermination, endpoints=%s",
utils.EndpointsToString(endpoints))
- _, err := cli.clientManager.NotifyClientTermination(ctx,
endpoints, request, cli.opts.timeout)
- if err != nil {
- cli.log.Errorf("failed to notify client termination,
endpoints=%s, error=%v", utils.EndpointsToString(endpoints), err)
- }
+ wg.Add(1)
+ go func(endpoints *v2.Endpoints) {
+ defer wg.Done()
+ cli.log.Infof("start notifyClientTermination,
endpoints=%s", utils.EndpointsToString(endpoints))
+ _, err :=
cli.clientManager.NotifyClientTermination(ctx, endpoints, request,
cli.opts.timeout)
+ if err != nil {
+ cli.log.Errorf("failed to notify client
termination, endpoints=%s, error=%v", utils.EndpointsToString(endpoints), err)
+ }
+ }(endpoints)
}
+ wg.Wait()
}
func (cli *defaultClient) GracefulStop() error {
diff --git a/golang/process_queue.go b/golang/process_queue.go
index a73aff03..2bc77662 100644
--- a/golang/process_queue.go
+++ b/golang/process_queue.go
@@ -19,6 +19,7 @@ package golang
import (
"context"
+ "errors"
"math"
"time"
@@ -45,11 +46,24 @@ const (
FORWARD_FIFO_MESSAGE_TO_DLQ_FAILURE_BACKOFF_DELAY time.Duration =
time.Second
ACK_MESSAGE_FAILURE_BACKOFF_DELAY time.Duration =
time.Second
CHANGE_INVISIBLE_DURATION_FAILURE_BACKOFF_DELAY time.Duration =
time.Second
- RECEIVING_FLOW_CONTROL_BACKOFF_DELAY time.Duration =
time.Millisecond * 20
RECEIVING_FAILURE_BACKOFF_DELAY time.Duration =
time.Second
RECEIVING_BACKOFF_DELAY_WHEN_CACHE_IS_FULL time.Duration =
time.Second
)
+// Consecutive reception failures after which the failure is logged again at
+// WARN level; in between, retries are only logged at DEBUG, so a sustained
+// outage does not produce one ERROR per second per queue.
+const RECEPTION_FAILURE_LOG_PERIOD = 30
+
+// Max attempts of the settlement chains (ack, change invisible duration,
forward
+// to DLQ) before the message is given up; every retry waits for the chain's
+// failure back-off delay, so a message is retried for a few seconds at most.
+const (
+ MAX_ACK_MESSAGE_ATTEMPTS = 3
+ MAX_CHANGE_INVISIBLE_DURATION_ATTEMPTS = 3
+ MAX_FORWARD_MESSAGE_TO_DLQ_ATTEMPTS = 3
+)
+
type defaultProcessQueue struct {
consumer *defaultPushConsumer
dropped atomic.Bool
@@ -63,6 +77,8 @@ type defaultProcessQueue struct {
receivedMessagesQuantity atomic.Int64
activityNanoTime atomic.Int64
cacheFullNanoTime atomic.Int64
+
+ consecutiveReceptionFailures atomic.Int64
}
func (dpq *defaultProcessQueue) discardFifoMessage(mv *MessageView) {
@@ -85,7 +101,7 @@ func (dpq *defaultProcessQueue) eraseFifoMessage(mv
*MessageView, result Consume
nextAttemptDelay := utils.GetNextAttemptDelay(retryPolicy,
int(attempt))
mv.deliveryAttempt += 1
attempt = mv.deliveryAttempt
- dpq.consumer.cli.log.Debugf("Prepare to redeliver the fifo
message because of the consumption failure, maxAttempt={},"+
+ dpq.consumer.cli.log.Debugf("Prepare to redeliver the fifo
message because of the consumption failure, maxAttempt=%d,"+
" attempt=%d, mq=%s, messageId=%s, nextAttemptDelay=%v,
clientId=%s", maxAttempts, attempt, dpq.mqstr,
messageId, nextAttemptDelay, clientId)
service.consumeWithDuration(mv, nextAttemptDelay, func(result0
ConsumerResult, err0 error) {
@@ -139,8 +155,8 @@ func (dpq *defaultProcessQueue)
forwardToDeadLetterQueue0(mv *MessageView, attem
ctx := context.Background()
resp, err := dpq.consumer.forwardMessageToDeadLetterQueue0(ctx, mv)
if err != nil {
- dpq.consumer.cli.log.Errorf("Exception raised while
acknowledging message, clientId=%s, consumerGroup=%s, "+
- "would attempt to re-ack later, attempt=%d,
messageId=%s, mq=%s, endpoints=%v, err=%w", clientId,
+ dpq.consumer.cli.log.Errorf("Exception raised while forwarding
message to dead letter queue, "+
+ "would attempt to re-forward later, attempt=%d,
messageId=%s, mq=%s, endpoints=%v, err=%v", clientId,
consumerGroup, attempt, messageId, dpq.mqstr,
endpoints, err)
dpq.forwardToDeadLetterQueueLater(mv, 1+attempt, callback)
return
@@ -148,6 +164,15 @@ func (dpq *defaultProcessQueue)
forwardToDeadLetterQueue0(mv *MessageView, attem
requestId := utils.GetRequestID(ctx)
status := resp.GetStatus()
code := status.GetCode()
+ if code == v2.Code_INVALID_RECEIPT_HANDLE {
+ // The receipt handle has expired, no retry can ever succeed.
+ dpq.consumer.cli.log.Errorf("Failed to forward message to dead
letter queue due to the invalid receipt handle, forgive to retry, "+
+ "clientId=%s, consumerGroup=%s, messageId=%s,
attempt=%d, mq=%s, endpoints=%v, "+
+ "requestId=%s, status message=[%s]", clientId,
consumerGroup, messageId, attempt, dpq.mqstr,
+ endpoints, requestId, status.GetMessage())
+ callback(&ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST),
Message: code.String()})
+ return
+ }
// Log failure and retry later.
if code != v2.Code_OK {
dpq.consumer.cli.log.Errorf("Failed to forward message to dead
letter queue, would attempt to re-forward later, "+
@@ -161,7 +186,7 @@ func (dpq *defaultProcessQueue)
forwardToDeadLetterQueue0(mv *MessageView, attem
// Log retries.
if attempt > 1 {
dpq.consumer.cli.log.Infof("Re-forward message to dead letter
queue successfully, clientId=%s, consumerGroup=%s "+
- "messageId={}, attempt={}, mq={}, endpoints={},
requestId={}", clientId, consumerGroup,
+ "messageId=%s, attempt=%d, mq=%s, endpoints=%v,
requestId=%s", clientId, consumerGroup,
messageId, attempt, dpq.mqstr, endpoints, requestId)
return
}
@@ -173,14 +198,22 @@ func (dpq *defaultProcessQueue)
forwardToDeadLetterQueue0(mv *MessageView, attem
func (dpq *defaultProcessQueue) forwardToDeadLetterQueueLater(mv *MessageView,
attempt int, callback func(error)) {
clientId := dpq.consumer.cli.clientID
messageId := mv.messageId
- time.AfterFunc(ACK_MESSAGE_FAILURE_BACKOFF_DELAY, func() {
+ if attempt > MAX_FORWARD_MESSAGE_TO_DLQ_ATTEMPTS {
+ dpq.forgiveSettlement("forward message to dead letter queue",
attempt, messageId, callback)
+ return
+ }
+ time.AfterFunc(FORWARD_FIFO_MESSAGE_TO_DLQ_FAILURE_BACKOFF_DELAY,
func() {
defer func() {
if err := recover(); err != nil {
- dpq.consumer.cli.log.Errorf("[Bug] Failed to
schedule message change invisible duration request, mq=%s, messageId=%s, "+
+ dpq.consumer.cli.log.Errorf("[Bug] Failed to
schedule message forward to dead letter queue request, mq=%s, messageId=%s, "+
"clientId=%s", dpq.mqstr, messageId,
clientId)
dpq.forwardToDeadLetterQueueLater(mv,
1+attempt, callback)
}
}()
+ if err := dpq.settlementRetryGuard("forward message to dead
letter queue", messageId); err != nil {
+ callback(err)
+ return
+ }
dpq.forwardToDeadLetterQueue0(mv, attempt, callback)
})
}
@@ -214,38 +247,33 @@ func (dpq *defaultProcessQueue)
changeInvisibleDuration(mv *MessageView, duratio
endpoints := mv.endpoints
ctx := context.Background()
- resp, err := dpq.consumer.changeInvisibleDuration0(ctx, mv, duration)
- if err != nil {
- dpq.consumer.cli.log.Errorf("Exception raised while changing
invisible duration, would retry later, clientId=%s, consumerGroup=%s,
messageId=%s, mq=%s, endpoints=%v, err=%w",
- clientId, consumerGroup, messageId, dpq.mqstr,
endpoints, err)
- dpq.changeInvisibleDurationLater(mv, duration, 1+attempt,
callback)
- return
- }
- requestId := utils.GetRequestID(ctx)
- status := resp.GetStatus()
- code := status.GetCode()
- if code == v2.Code_INVALID_RECEIPT_HANDLE {
+ _, err := dpq.consumer.changeInvisibleDuration0(ctx, mv, duration)
+ // changeInvisibleDuration0 converts a non-OK status into an error, so
an expired
+ // receipt handle surfaces through err instead of through a response
status:
+ // the abandon branch must run before the generic failure handling,
otherwise
+ // the expired handle is renewed once a second forever.
+ if rpcErr, ok := AsErrRpcStatus(err); ok && rpcErr.GetCode() ==
int32(v2.Code_INVALID_RECEIPT_HANDLE) {
dpq.consumer.cli.log.Errorf("Failed to change invisible
duration due to the invalid receipt handle, forgive to retry, "+
- "clientId=%s, consumerGroup=%s, messageId=%s,
attempt=%d, mq=%s, endpoints=%v, "+
- "requestId=%s, status message=[%s]", clientId,
consumerGroup, messageId, attempt, dpq.mqstr,
- endpoints, requestId, status.GetMessage())
- callback(&ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST),
Message: code.String()})
+ "clientId=%s, consumerGroup=%s, messageId=%s,
attempt=%d, mq=%s, endpoints=%v, status message=[%s]",
+ clientId, consumerGroup, messageId, attempt, dpq.mqstr,
endpoints, rpcErr.GetMessage())
+ callback(&ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST),
Message: v2.Code_INVALID_RECEIPT_HANDLE.String()})
return
}
- // Log failure and retry later.
- if code != v2.Code_OK {
- dpq.consumer.cli.log.Errorf("Failed to change invisible
duration, would retry later, "+
- " clientId=%s, consumerGroup=%s, messageId=%s,
attempt=%d, mq=%s, endpoints=%v, requestId=%s, status message=[%s]", clientId,
consumerGroup, messageId, attempt, dpq.mqstr,
- endpoints, requestId, status.GetMessage())
+ if err != nil {
+ dpq.consumer.cli.log.Errorf("Exception raised while changing
invisible duration, would retry later, clientId=%s, consumerGroup=%s,
messageId=%s, mq=%s, endpoints=%v, err=%v",
+ clientId, consumerGroup, messageId, dpq.mqstr,
endpoints, err)
dpq.changeInvisibleDurationLater(mv, duration, 1+attempt,
callback)
return
}
+ // A nil error implies the server returned Code_OK, any other status was
+ // converted into the error handled above.
+ requestId := utils.GetRequestID(ctx)
// Set result if succeed in changing invisible time.
callback(nil)
// Log retries.
if attempt > 1 {
dpq.consumer.cli.log.Infof("Finally, change invisible duration
successfully, clientId=%s, consumerGroup=%s "+
- "messageId={}, attempt={}, mq={}, endpoints={},
requestId={}", clientId, consumerGroup,
+ "messageId=%s, attempt=%d, mq=%s, endpoints=%v,
requestId=%s", clientId, consumerGroup,
messageId, attempt, dpq.mqstr, endpoints, requestId)
return
}
@@ -257,6 +285,10 @@ func (dpq *defaultProcessQueue) changeInvisibleDuration(mv
*MessageView, duratio
func (dpq *defaultProcessQueue) changeInvisibleDurationLater(mv *MessageView,
duration time.Duration, attempt int, callback func(error)) {
clientId := dpq.consumer.cli.clientID
messageId := mv.messageId
+ if attempt > MAX_CHANGE_INVISIBLE_DURATION_ATTEMPTS {
+ dpq.forgiveSettlement("change invisible duration", attempt,
messageId, callback)
+ return
+ }
time.AfterFunc(CHANGE_INVISIBLE_DURATION_FAILURE_BACKOFF_DELAY, func() {
defer func() {
if err := recover(); err != nil {
@@ -265,6 +297,10 @@ func (dpq *defaultProcessQueue)
changeInvisibleDurationLater(mv *MessageView, du
dpq.changeInvisibleDurationLater(mv, duration,
1+attempt, callback)
}
}()
+ if err := dpq.settlementRetryGuard("change invisible duration",
messageId); err != nil {
+ callback(err)
+ return
+ }
dpq.changeInvisibleDuration(mv, duration, attempt, callback)
})
}
@@ -283,7 +319,7 @@ func (dpq *defaultProcessQueue) ackMessage0(mv
*MessageView, attempt int, callba
resp, err := dpq.consumer.ack0(ctx, mv)
if err != nil {
dpq.consumer.cli.log.Errorf("Exception raised while
acknowledging message, clientId=%s, consumerGroup=%s, "+
- "would attempt to re-ack later, attempt=%d,
messageId=%s, mq=%s, endpoints=%v, err=%w", clientId,
+ "would attempt to re-ack later, attempt=%d,
messageId=%s, mq=%s, endpoints=%v, err=%v", clientId,
consumerGroup, attempt, messageId, dpq.mqstr,
endpoints, err)
dpq.ackMessageLater(mv, 1+attempt, callback)
return
@@ -312,7 +348,7 @@ func (dpq *defaultProcessQueue) ackMessage0(mv
*MessageView, attempt int, callba
// Log retries.
if attempt > 1 {
dpq.consumer.cli.log.Infof("Finally, ack message successfully,
clientId=%s, consumerGroup=%s "+
- "messageId={}, attempt={}, mq={}, endpoints={},
requestId={}", clientId, consumerGroup,
+ "messageId=%s, attempt=%d, mq=%s, endpoints=%v,
requestId=%s", clientId, consumerGroup,
messageId, attempt, dpq.mqstr, endpoints, requestId)
return
}
@@ -324,18 +360,51 @@ func (dpq *defaultProcessQueue) ackMessage0(mv
*MessageView, attempt int, callba
func (dpq *defaultProcessQueue) ackMessageLater(mv *MessageView, attempt int,
callback func(error)) {
clientId := dpq.consumer.cli.clientID
messageId := mv.messageId
+ if attempt > MAX_ACK_MESSAGE_ATTEMPTS {
+ dpq.forgiveSettlement("ack message", attempt, messageId,
callback)
+ return
+ }
time.AfterFunc(ACK_MESSAGE_FAILURE_BACKOFF_DELAY, func() {
defer func() {
if err := recover(); err != nil {
- dpq.consumer.cli.log.Errorf("[Bug] Failed to
schedule message change invisible duration request, mq=%s, messageId=%s, "+
+ dpq.consumer.cli.log.Errorf("[Bug] Failed to
schedule message acknowledgement request, mq=%s, messageId=%s, "+
"clientId=%s", dpq.mqstr, messageId,
clientId)
dpq.ackMessageLater(mv, 1+attempt, callback)
}
}()
+ if err := dpq.settlementRetryGuard("ack message", messageId);
err != nil {
+ callback(err)
+ return
+ }
dpq.ackMessage0(mv, attempt, callback)
})
}
+// forgiveSettlement reports the final give-up of a settlement retry chain; the
+// callback evicts the cached message so the queue cache does not leak.
+func (dpq *defaultProcessQueue) forgiveSettlement(operation string, attempt
int, messageId string, callback func(error)) {
+ dpq.consumer.cli.log.Errorf("%s failed finally, run out of attempt
times, forgive to retry, attempt=%d, messageId=%s, mq=%s, clientId=%s",
+ operation, attempt, messageId, dpq.mqstr,
dpq.consumer.cli.clientID)
+ callback(&ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST), Message:
operation + " gave up after retries"})
+}
+
+// settlementRetryGuard stops the scheduled settlement retries once the queue
was
+// dropped or the consumer is no longer running; the returned error, if any, is
+// handed to the caller's callback so the cached message is still evicted.
+func (dpq *defaultProcessQueue) settlementRetryGuard(operation string,
messageId string) error {
+ if dpq.dropped.Load() {
+ dpq.consumer.cli.log.Infof("Stop to retry %s because the
process queue has been dropped, mq=%s, messageId=%s, clientId=%s",
+ operation, dpq.mqstr, messageId,
dpq.consumer.cli.clientID)
+ return &ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST), Message:
"process queue has been dropped"}
+ }
+ if !dpq.consumer.isRunning() {
+ dpq.consumer.cli.log.Infof("Stop to retry %s because the
consumer is not running, mq=%s, messageId=%s, clientId=%s",
+ operation, dpq.mqstr, messageId,
dpq.consumer.cli.clientID)
+ return &ErrRpcStatus{Code: int32(v2.Code_BAD_REQUEST), Message:
"consumer is not running"}
+ }
+ return nil
+}
+
func (dpq *defaultProcessQueue) getMessageQueue() *v2.MessageQueue {
return dpq.mq
}
@@ -410,7 +479,7 @@ func (dpq *defaultProcessQueue) isCacheFull() bool {
cacheMessageBytesThresholdPerQueue :=
int64(dpq.consumer.cacheMessageBytesThresholdPerQueue())
actualCachedMessagesBytes := dpq.cachedMessagesBytes.Load()
if cacheMessageBytesThresholdPerQueue <= actualCachedMessagesBytes {
- dpq.consumer.cli.log.Warnf("Process queue total cached messages
memory exceeds the threshold, threshold={} bytes, actual={} bytes, mq={},
clientId={}",
+ dpq.consumer.cli.log.Warnf("Process queue total cached messages
memory exceeds the threshold, threshold=%d bytes, actual=%d bytes, mq=%s,
clientId=%s",
cacheMessageBytesThresholdPerQueue,
actualCachedMessagesBytes, dpq.mqstr, clientId)
dpq.cacheFullNanoTime.Store(time.Now().UnixNano())
return true
@@ -433,7 +502,7 @@ func (dpq *defaultProcessQueue)
receiveMessageImmediatelyWithAttemptId(attemptId
defer func() {
if err := recover(); err != nil {
dpq.consumer.cli.log.Errorf("Exception raised during
message reception, mq=%s, clientId=%s, err=%v", dpq.mqstr, clientId, err)
- dpq.onReceiveMessageException(err, attemptId)
+ dpq.onReceiveMessageException(attemptId)
}
}()
@@ -450,36 +519,63 @@ func (dpq *defaultProcessQueue)
receiveMessageImmediatelyWithAttemptId(attemptId
timeout := longPollingTimeout + dpq.consumer.cli.opts.timeout
go func() {
- mvs, err := dpq.consumer.receiveMessage(context.TODO(),
request, dpq.mq, timeout)
+ mvs, err :=
dpq.consumer.receiveMessage(dpq.consumer.receiveCtx, request, dpq.mq, timeout)
duration := time.Since(startTime)
if err == nil {
+ dpq.consecutiveReceptionFailures.Store(0)
messageCommons := make([]*MessageCommon, 0, len(mvs))
for _, mv := range mvs {
messageCommons = append(messageCommons,
mv.GetMessageCommon())
}
dpq.consumer.cli.doAfter(MessageHookPoints_RECEIVE,
messageCommons, duration, MessageHookPointsStatus_OK)
dpq.onReceiveMessageResult(mvs)
+ return
+ }
+ nextAttemptId := ""
+ if status.Code(err) == codes.DeadlineExceeded {
+ nextAttemptId = request.GetAttemptId()
+ }
+ rpcErr, isRpcErr := AsErrRpcStatus(err)
+ isNoNewMessage := isRpcErr && rpcErr.GetCode() ==
int32(v2.Code_MESSAGE_NOT_FOUND)
+ // An empty long polling result is still a completed reception,
so it has
+ // to close the hook opened before the request was sent.
+ hookStatus := MessageHookPointsStatus_OK
+ if isNoNewMessage {
+ dpq.consecutiveReceptionFailures.Store(0)
+ dpq.consumer.cli.log.Debugf("No new message, mq=%s,
endpoints=%v, clientId=%s",
+ dpq.mqstr, endpoints, clientId)
+ } else if errors.Is(err, context.Canceled) {
+ hookStatus = MessageHookPointsStatus_ERROR
+ // Cancellation only happens while the consumer is
stopping and the
+ // queue will not receive again, so neither log it as a
reception
+ // failure nor schedule another receive for it.
+ dpq.consumer.cli.log.Infof("Receive request was
canceled, mq=%s, endpoints=%v, attemptId=%s, clientId=%s",
+ dpq.mqstr, endpoints, request.GetAttemptId(),
clientId)
} else {
- nextAttemptId := ""
- if status.Code(err) == codes.DeadlineExceeded {
- nextAttemptId = request.GetAttemptId()
- }
- rpcErr, isRpcErr := AsErrRpcStatus(err)
- isNoNewMessage := isRpcErr && rpcErr.GetCode() ==
int32(v2.Code_MESSAGE_NOT_FOUND)
- // An empty long polling result is still a completed
reception, so it has
- // to close the hook opened before the request was sent.
- hookStatus := MessageHookPointsStatus_OK
- if isNoNewMessage {
- dpq.consumer.cli.log.Debugf("No new message,
mq=%s, endpoints=%v, clientId=%s",
- dpq.mqstr, endpoints, clientId)
- } else {
- hookStatus = MessageHookPointsStatus_ERROR
+ hookStatus = MessageHookPointsStatus_ERROR
+ // Reception retries forever by design, so a sustained
outage must not
+ // produce one ERROR per second per queue: only the
first failure and
+ // every RECEPTION_FAILURE_LOG_PERIOD-th consecutive
one are logged at
+ // ERROR/WARN, the retries in between stay at DEBUG.
+ failures := dpq.consecutiveReceptionFailures.Add(1)
+ switch {
+ case failures == 1:
dpq.consumer.cli.log.Errorf("Exception raised
during message reception, mq=%s, endpoints=%v, attemptId=%s, "+
"nextAttemptId=%s, clientId=%s,
err=%v", dpq.mqstr, endpoints, request.GetAttemptId(), nextAttemptId,
clientId, err)
+ case failures%RECEPTION_FAILURE_LOG_PERIOD == 0:
+ dpq.consumer.cli.log.Warnf("Message reception
keeps failing, consecutiveFailures=%d, mq=%s, endpoints=%v, "+
+ "attemptId=%s, nextAttemptId=%s,
clientId=%s, err=%v", failures, dpq.mqstr, endpoints,
+ request.GetAttemptId(), nextAttemptId,
clientId, err)
+ default:
+ dpq.consumer.cli.log.Debugf("Exception raised
during message reception, mq=%s, endpoints=%v, attemptId=%s, "+
+ "nextAttemptId=%s, clientId=%s,
err=%v", dpq.mqstr, endpoints, request.GetAttemptId(), nextAttemptId,
+ clientId, err)
}
- dpq.consumer.cli.doAfter(MessageHookPoints_RECEIVE,
make([]*MessageCommon, 0), duration, hookStatus)
- dpq.onReceiveMessageException(err, nextAttemptId)
+ }
+ dpq.consumer.cli.doAfter(MessageHookPoints_RECEIVE,
make([]*MessageCommon, 0), duration, hookStatus)
+ if !errors.Is(err, context.Canceled) {
+ dpq.onReceiveMessageException(nextAttemptId)
}
}()
dpq.receptionTimes.Inc()
@@ -493,16 +589,11 @@ func (dpq *defaultProcessQueue) getReceptionBatchSize()
int32 {
return int32(math.Min(bufferSize,
float64(dpq.consumer.pcSettings.receiveBatchSize)))
}
-func (dpq *defaultProcessQueue) onReceiveMessageException(t any, attemptId
string) {
- duration := RECEIVING_FLOW_CONTROL_BACKOFF_DELAY
- error, ok := t.(error)
- if ok {
- rpcError, ok := AsErrRpcStatus(error)
- if ok && rpcError.GetCode() == int32(v2.Code_TOO_MANY_REQUESTS)
{
- duration = RECEIVING_FAILURE_BACKOFF_DELAY
- }
- }
- dpq.receiveMessageLater(duration, attemptId)
+func (dpq *defaultProcessQueue) onReceiveMessageException(attemptId string) {
+ // A short flow-control back-off would hammer an overloaded proxy at
dozens of
+ // polls per second, so every reception failure waits a full second
before
+ // polling again.
+ dpq.receiveMessageLater(RECEIVING_FAILURE_BACKOFF_DELAY, attemptId)
}
func (dpq *defaultProcessQueue) receiveMessageLater(duration time.Duration,
attemptId string) {
@@ -510,10 +601,10 @@ func (dpq *defaultProcessQueue)
receiveMessageLater(duration time.Duration, atte
defer func() {
if err := recover(); err != nil {
dpq.consumer.cli.log.Errorf("[Bug] Failed to
schedule message receiving request, mq=%s, clientId=%s, err=%v", dpq.mqstr,
dpq.consumer.cli.clientID, err)
- dpq.onReceiveMessageException(err, attemptId)
+ dpq.onReceiveMessageException(attemptId)
}
}()
- dpq.consumer.cli.log.Infof("Try to receive message later,
mq=%s, delay=%v, clientId=%s", dpq.mqstr, duration, dpq.consumer.cli.clientID)
+ dpq.consumer.cli.log.Debugf("Try to receive message later,
mq=%s, delay=%v, clientId=%s", dpq.mqstr, duration, dpq.consumer.cli.clientID)
dpq.receiveMessageWithAttemptId(attemptId)
})
}
diff --git a/golang/process_queue_test.go b/golang/process_queue_test.go
index c6c4576d..66f71597 100644
--- a/golang/process_queue_test.go
+++ b/golang/process_queue_test.go
@@ -19,6 +19,7 @@ package golang
import (
"context"
+ "errors"
"io"
"sync/atomic"
"testing"
@@ -124,3 +125,262 @@ func
TestProcessQueueEmptyLongPollingBalancesInflightReceiveCount(t *testing.T)
"wait out requestTimeout + longPollingTimeout before
giving up", count, receptions.Load())
}
}
+
+// newSettlementTestProcessQueue builds a push consumer whose RPCs are served
by
+// a mock client manager, plus one process queue to drive the settlement chains
+// (ack, change invisible duration, forward to DLQ) against it.
+func newSettlementTestProcessQueue(t *testing.T) (*defaultPushConsumer,
*defaultProcessQueue, *MockClientManager) {
+ t.Helper()
+ consumer, err := NewPushConsumer(&Config{
+ Endpoint: fakeAddress,
+ ConsumerGroup: MOCK_GROUP,
+ Credentials: &credentials.SessionCredentials{},
+ },
+
WithPushSubscriptionExpressions(map[string]*FilterExpression{MOCK_TOPIC:
SUB_ALL}),
+ WithPushMessageListener(&FuncMessageListener{
+ Consume: func(*MessageView) ConsumerResult { return
SUCCESS },
+ }),
+ )
+ if err != nil {
+ t.Fatal(err)
+ }
+ pc := consumer.(*defaultPushConsumer)
+ manager := NewMockClientManager(gomock.NewController(t))
+ pc.cli.clientManager = manager
+ pc.cli.log = zap.NewNop().Sugar()
+ pc.cli.inited.Store(true)
+
+ mq := &v2.MessageQueue{
+ Topic: &v2.Resource{Name: MOCK_TOPIC},
+ Id: 0,
+ Broker: &v2.Broker{Name: "broker", Endpoints:
fakeEndpoints()},
+ AcceptMessageTypes: []v2.MessageType{v2.MessageType_NORMAL},
+ }
+ return pc, newDefaultProcessQueue(pc, utils.ParseMessageQueue2Str(mq),
mq, SUB_ALL), manager
+}
+
+func testMessageView() *MessageView {
+ return &MessageView{
+ messageId: "message-1",
+ topic: MOCK_TOPIC,
+ body: []byte("body"),
+ endpoints: fakeEndpoints(),
+ ReceiptHandle: "receipt-handle-0",
+ }
+}
+
+// An expired receipt handle can never be renewed, so the chain must abandon at
+// once instead of retrying once a second forever.
+func TestChangeInvisibleDurationForgivesInvalidReceiptHandle(t *testing.T) {
+ _, dpq, manager := newSettlementTestProcessQueue(t)
+ defer dpq.drop()
+
+ var calls atomic.Int32
+ manager.EXPECT().ChangeInvisibleDuration(gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.ChangeInvisibleDurationRequest, time.Duration)
(*v2.ChangeInvisibleDurationResponse, error) {
+ calls.Add(1)
+ return &v2.ChangeInvisibleDurationResponse{
+ Status: &v2.Status{Code:
v2.Code_INVALID_RECEIPT_HANDLE, Message:
v2.Code_INVALID_RECEIPT_HANDLE.String()},
+ }, nil
+ }).Times(1)
+
+ var got error
+ dpq.changeInvisibleDuration(testMessageView(), time.Second, 1, func(err
error) { got = err })
+ if got == nil {
+ t.Fatal("expected the callback to report the abandoned renewal")
+ }
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("expected the renewal to be abandoned without a retry,
got %d attempts", got)
+ }
+}
+
+func TestChangeInvisibleDurationRetriesAreCapped(t *testing.T) {
+ _, dpq, manager := newSettlementTestProcessQueue(t)
+ defer dpq.drop()
+
+ var calls atomic.Int32
+ manager.EXPECT().ChangeInvisibleDuration(gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.ChangeInvisibleDurationRequest, time.Duration)
(*v2.ChangeInvisibleDurationResponse, error) {
+ calls.Add(1)
+ return nil, errors.New("rpc unavailable")
+ }).Times(MAX_CHANGE_INVISIBLE_DURATION_ATTEMPTS)
+
+ done := make(chan error, 1)
+ go func() {
+ dpq.changeInvisibleDuration(testMessageView(), time.Second, 1,
func(err error) { done <- err })
+ }()
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("expected the callback to report the give-up")
+ }
+ case <-time.After(15 * time.Second):
+ t.Fatal("the callback was never invoked after the retry budget
was exhausted")
+ }
+ if got := calls.Load(); got != MAX_CHANGE_INVISIBLE_DURATION_ATTEMPTS {
+ t.Fatalf("expected exactly %d attempts, got %d",
MAX_CHANGE_INVISIBLE_DURATION_ATTEMPTS, got)
+ }
+}
+
+// A dropped process queue must not keep issuing settlement RPCs for its cached
+// messages; the callback still has to fire so the cache bookkeeping is undone.
+func TestAckMessageStopsWhenProcessQueueDropped(t *testing.T) {
+ _, dpq, manager := newSettlementTestProcessQueue(t)
+
+ var calls atomic.Int32
+ manager.EXPECT().AckMessage(gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.AckMessageRequest, time.Duration) (*v2.AckMessageResponse, error) {
+ calls.Add(1)
+ return nil, errors.New("rpc unavailable")
+ }).Times(1)
+
+ done := make(chan error, 1)
+ go func() {
+ dpq.ackMessage(testMessageView(), func(err error) { done <- err
})
+ }()
+ time.Sleep(100 * time.Millisecond)
+ dpq.drop()
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("expected the callback to report the abandoned
retry")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the retry guard never released the callback")
+ }
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("expected the retry to be abandoned after the first
attempt, got %d attempts", got)
+ }
+}
+
+func TestForwardToDeadLetterQueueForgivesInvalidReceiptHandle(t *testing.T) {
+ _, dpq, manager := newSettlementTestProcessQueue(t)
+ defer dpq.drop()
+
+ var calls atomic.Int32
+ manager.EXPECT().ForwardMessageToDeadLetterQueue(gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.ForwardMessageToDeadLetterQueueRequest, time.Duration)
(*v2.ForwardMessageToDeadLetterQueueResponse, error) {
+ calls.Add(1)
+ return &v2.ForwardMessageToDeadLetterQueueResponse{
+ Status: &v2.Status{Code:
v2.Code_INVALID_RECEIPT_HANDLE, Message:
v2.Code_INVALID_RECEIPT_HANDLE.String()},
+ }, nil
+ }).Times(1)
+
+ var got error
+ dpq.forwardToDeadLetterQueue(testMessageView(), func(err error) { got =
err })
+ if got == nil {
+ t.Fatal("expected the callback to report the abandoned forward")
+ }
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("expected the forward to be abandoned without a retry,
got %d attempts", got)
+ }
+}
+
+func TestForwardToDeadLetterQueueStopsWhenConsumerNotRunning(t *testing.T) {
+ pc, dpq, manager := newSettlementTestProcessQueue(t)
+
+ var calls atomic.Int32
+ manager.EXPECT().ForwardMessageToDeadLetterQueue(gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.ForwardMessageToDeadLetterQueueRequest, time.Duration)
(*v2.ForwardMessageToDeadLetterQueueResponse, error) {
+ calls.Add(1)
+ return nil, errors.New("rpc unavailable")
+ }).Times(1)
+
+ done := make(chan error, 1)
+ go func() {
+ dpq.forwardToDeadLetterQueue(testMessageView(), func(err error)
{ done <- err })
+ }()
+ time.Sleep(100 * time.Millisecond)
+ pc.cli.on.Store(false)
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("expected the callback to report the abandoned
retry")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("the retry guard never released the callback")
+ }
+ if got := calls.Load(); got != 1 {
+ t.Fatalf("expected the retry to be abandoned after the first
attempt, got %d attempts", got)
+ }
+}
+
+// A sustained settlement failure must end in the give-up: the ack chain stops
+// after MAX_ACK_MESSAGE_ATTEMPTS attempts instead of retrying once a second
+// forever.
+func TestAckMessageRetriesAreCapped(t *testing.T) {
+ _, dpq, manager := newSettlementTestProcessQueue(t)
+ defer dpq.drop()
+
+ var calls atomic.Int32
+ manager.EXPECT().AckMessage(gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.AckMessageRequest, time.Duration) (*v2.AckMessageResponse, error) {
+ calls.Add(1)
+ return nil, errors.New("rpc unavailable")
+ }).Times(MAX_ACK_MESSAGE_ATTEMPTS)
+
+ done := make(chan error, 1)
+ go func() {
+ dpq.ackMessage(testMessageView(), func(err error) { done <- err
})
+ }()
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("expected the callback to report the give-up")
+ }
+ case <-time.After(15 * time.Second):
+ t.Fatal("the callback was never invoked after the retry budget
was exhausted")
+ }
+ if got := calls.Load(); got != MAX_ACK_MESSAGE_ATTEMPTS {
+ t.Fatalf("expected exactly %d attempts, got %d",
MAX_ACK_MESSAGE_ATTEMPTS, got)
+ }
+}
+
+// hangUntilCanceledStream answers nothing: Recv blocks until the request
+// context is canceled, the way a long poll against a restarting proxy hangs.
+type hangUntilCanceledStream struct {
+ grpc.ClientStream
+ ctx context.Context
+}
+
+func (stream *hangUntilCanceledStream) Recv() (*v2.ReceiveMessageResponse,
error) {
+ <-stream.ctx.Done()
+ return nil, stream.ctx.Err()
+}
+
+// A poll hanging against a restarting proxy must not stretch the shutdown by
+// requestTimeout + longPollingTimeout: GracefulStop cancels the in-flight
+// receive requests, so the wait drains at once.
+func TestGracefulStopCancelsInflightReceiveRequest(t *testing.T) {
+ pc, dpq, manager := newSettlementTestProcessQueue(t)
+
+ manager.EXPECT().ReceiveMessage(gomock.Any(), gomock.Any(),
gomock.Any()).
+ DoAndReturn(func(ctx context.Context, _ *v2.Endpoints, _
*v2.ReceiveMessageRequest) (v2.MessagingService_ReceiveMessageClient, error) {
+ return &hangUntilCanceledStream{ctx: ctx}, nil
+ }).AnyTimes()
+ manager.EXPECT().shutdown().AnyTimes()
+
+ dpq.receiveMessageImmediately()
+ deadline := time.Now().Add(5 * time.Second)
+ for pc.inflightRequestCountInterceptor.getInflightReceiveRequestCount()
!= 1 && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond)
+ }
+ if got :=
pc.inflightRequestCountInterceptor.getInflightReceiveRequestCount(); got != 1 {
+ t.Fatalf("expected one in-flight receive request, got %d", got)
+ }
+
+ stopped := make(chan error, 1)
+ go func() { stopped <- pc.GracefulStop() }()
+ select {
+ case err := <-stopped:
+ if err != nil {
+ t.Fatalf("GracefulStop failed: %v", err)
+ }
+ case <-time.After(20 * time.Second):
+ t.Fatal("GracefulStop did not finish in time; the hanging poll
was not canceled and the shutdown " +
+ "waited out requestTimeout + longPollingTimeout")
+ }
+ if got :=
pc.inflightRequestCountInterceptor.getInflightReceiveRequestCount(); got != 0 {
+ t.Fatalf("inflight receive count is %d after GracefulStop; the
hanging poll was not canceled", got)
+ }
+}
diff --git a/golang/push_consumer.go b/golang/push_consumer.go
index 6cf13cfc..4a8e8111 100644
--- a/golang/push_consumer.go
+++ b/golang/push_consumer.go
@@ -19,6 +19,7 @@ package golang
import (
"context"
+ "errors"
"fmt"
"io"
"math"
@@ -69,6 +70,8 @@ type defaultPushConsumer struct {
consumptionErrorQuantity atomic.Int64
stopping atomic.Bool
+ receiveCtx context.Context
+ receiveCancel context.CancelFunc
inflightRequestCountInterceptor *defultInflightRequestCountInterceptor
pushConsumerExtension PushConsumerExtension
@@ -134,7 +137,7 @@ func (pc *defaultPushConsumer)
changeInvisibleDuration0(context context.Context,
duration := time.Since(watchTime)
messageHookPointsStatus := MessageHookPointsStatus_OK
if err != nil {
- pc.cli.log.Errorf("exception raised during message
acknowledgement, messageId=%s, endpoints=%v, requestId=%s",
messageView.GetMessageId(), endpoints, utils.GetRequestID(ctx))
+ pc.cli.log.Errorf("exception raised during message change
invisible duration, messageId=%s, endpoints=%v, requestId=%s",
messageView.GetMessageId(), endpoints, utils.GetRequestID(ctx))
} else if resp.GetStatus().GetCode() != v2.Code_OK {
pc.cli.log.Errorf("failed to change message invisible duration,
messageId=%s, endpoints=%v, code=%v, status message=[%s], requestId=%s",
messageView.GetMessageId(), endpoints, resp.GetStatus().GetCode(),
resp.GetStatus().GetMessage(), utils.GetRequestID(ctx))
err = &ErrRpcStatus{
@@ -257,6 +260,9 @@ func (pc *defaultPushConsumer) receiveMessage(ctx
context.Context, request *v2.R
}()
select {
case <-ctx.Done():
+ if errors.Is(ctx.Err(), context.Canceled) {
+ return nil, ctx.Err()
+ }
// timeout
return nil, fmt.Errorf("[error] CODE=DEADLINE_EXCEEDED")
case result := <-done:
@@ -353,6 +359,10 @@ var newPushConsumer = func(config *Config, opts
...PushConsumerOption) (*default
stopping: *atomic.NewBool(false),
inflightRequestCountInterceptor:
NewDefultInflightRequestCountInterceptor(),
}
+ // In-flight long polling receive requests are canceled on
GracefulStop, so
+ // the shutdown does not wait out the requestTimeout +
longPollingTimeout of
+ // every hanging poll.
+ pc.receiveCtx, pc.receiveCancel =
context.WithCancel(context.Background())
pc.pushConsumerExtension = pc
pc.cli.initTopics = make([]string, 0)
pcOpts.subscriptionExpressions.Range(func(key, value interface{}) bool {
@@ -430,8 +440,13 @@ func (pc *defaultPushConsumer) scanAssignments() {
filterExpression := value.(*FilterExpression)
newest, err := pc.cli.queryAssignments(context.TODO(), topic,
pc.groupName, pc.cli.opts.timeout)
if err != nil {
- pc.cli.log.Errorf("Exception raised while scanning the
assignments, topic=%s, clientId=%s, err=%v",
+ // A failed scan does not mean the server has no
assignments: syncing the
+ // nil result here would drop every working process
queue and stall
+ // consumption for the whole outage. Keep the current
assignments until
+ // the next successful scan.
+ pc.cli.log.Errorf("Exception raised while scanning the
assignments, keep current assignments, topic=%s, clientId=%s, err=%v",
topic, pc.cli.clientID, err)
+ return true
}
val, _ := pc.cacheAssignments.Load(topic)
var existed *[]*v2.Assignment
@@ -443,7 +458,7 @@ func (pc *defaultPushConsumer) scanAssignments() {
pc.cli.log.Infof("Acquired empty assignments
from remote, would scan later, topic=%s, clientId=%s", topic, pc.cli.clientID)
return true
}
- pc.cli.log.Infof("Attention!!! acquired empty
assignments from remote, but existed assignments is not empty, topic=%s,
clientId=%d",
+ pc.cli.log.Infof("Attention!!! acquired empty
assignments from remote, but existed assignments is not empty, topic=%s,
clientId=%s",
topic, pc.cli.clientID)
}
if !utils.CompareAssignments(newest, existed) {
@@ -495,7 +510,7 @@ func (pc *defaultPushConsumer) syncProcessQueue(topic
string, assignments *[]*v2
}
optionalProcessQueue := pc.createProcessQueue(mqs, mq,
filterExpression)
if optionalProcessQueue != nil {
- pc.cli.log.Infof("Start to fetch message from remote,
mq=%s, clientId={}", mqs, pc.cli.clientID)
+ pc.cli.log.Infof("Start to fetch message from remote,
mq=%s, clientId=%s", mqs, pc.cli.clientID)
optionalProcessQueue.fetchMessageImmediately()
}
}
@@ -529,6 +544,9 @@ func (pc *defaultPushConsumer) dropProcessQueue(mqstr
utils.MessageQueueStr) {
func (pc *defaultPushConsumer) GracefulStop() error {
// step 1 and 2
pc.stopping.Store(true)
+ // Cancel the long polling receive requests that are in flight, so the
wait
+ // below is not stretched by a poll hanging against a restarting proxy.
+ pc.receiveCancel()
// step 3
pc.cli.log.Infof("Waiting for the inflight receive requests to be
finished, clientId=%s", pc.cli.clientID)
diff --git a/golang/push_consumer_test.go b/golang/push_consumer_test.go
index 77c9039f..3e8436f5 100644
--- a/golang/push_consumer_test.go
+++ b/golang/push_consumer_test.go
@@ -18,9 +18,12 @@
package golang
import (
+ "fmt"
"testing"
"time"
+ "github.com/golang/mock/gomock"
+
v2 "github.com/apache/rocketmq-clients/golang/v5/protocol/v2"
)
@@ -276,3 +279,21 @@ func TestPredefinedConsumerResults(t *testing.T) {
t.Errorf("FAILURE.Type = %v, want %v", FAILURE.Type,
ConsumerResultTypeFailure)
}
}
+
+func TestScanAssignmentsKeepsQueuesWhenScanFails(t *testing.T) {
+ pc, dpq, manager := newSettlementTestProcessQueue(t)
+ pc.processQueueTable.Store(dpq.mqstr, []interface{}{dpq.mq, dpq})
+ assignments := &[]*v2.Assignment{{MessageQueue: dpq.mq}}
+ pc.cacheAssignments.Store(MOCK_TOPIC, assignments)
+ manager.EXPECT().QueryAssignments(gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any()).
+ Return(nil, fmt.Errorf("proxy down")).AnyTimes()
+
+ pc.scanAssignments()
+
+ if _, ok := pc.processQueueTable.Load(dpq.mqstr); !ok {
+ t.Fatalf("process queue was dropped after a failed assignment
scan; an RPC error must not be treated as empty assignments")
+ }
+ if cached, _ := pc.cacheAssignments.Load(MOCK_TOPIC); cached !=
assignments {
+ t.Fatalf("cached assignments were replaced after a failed scan")
+ }
+}
diff --git a/golang/simple_consumer.go b/golang/simple_consumer.go
index ef7b9ee4..beed8f75 100644
--- a/golang/simple_consumer.go
+++ b/golang/simple_consumer.go
@@ -118,7 +118,7 @@ func (sc *defaultSimpleConsumer)
changeInvisibleDuration0(messageView *MessageVi
duration := time.Since(watchTime)
messageHookPointsStatus := MessageHookPointsStatus_OK
if err != nil {
- sc.cli.log.Errorf("exception raised during message
acknowledgement, messageId=%s, endpoints=%v, requestId=%s",
messageView.GetMessageId(), endpoints, utils.GetRequestID(ctx))
+ sc.cli.log.Errorf("exception raised during message change
invisible duration, messageId=%s, endpoints=%v, requestId=%s",
messageView.GetMessageId(), endpoints, utils.GetRequestID(ctx))
} else if resp.GetStatus().GetCode() != v2.Code_OK {
sc.cli.log.Errorf("failed to change message invisible duration,
messageId=%s, endpoints=%v, code=%v, status message=[%s], requestId=%s",
messageView.GetMessageId(), endpoints, resp.GetStatus().GetCode(),
resp.GetStatus().GetMessage(), utils.GetRequestID(ctx))
err = &ErrRpcStatus{
@@ -138,7 +138,9 @@ func (sc *defaultSimpleConsumer)
changeInvisibleDuration(messageView *MessageVie
return fmt.Errorf("changeInvisibleDuration failed, err = the
message is nil")
}
resp, err := sc.changeInvisibleDuration0(messageView, invisibleDuration)
- if resp != nil {
+ // A failed request carries no new receipt handle: adopting the empty
one from
+ // the response would make every following call fail with
INVALID_RECEIPT_HANDLE.
+ if err == nil && resp != nil {
messageView.ReceiptHandle = resp.ReceiptHandle
}
return err
@@ -503,9 +505,13 @@ func (sc *defaultSimpleConsumer) Ack(ctx context.Context,
messageView *MessageVi
}
if resp.GetStatus().GetCode() != v2.Code_OK {
messageHookPointsStatus = MessageHookPointsStatus_ERROR
+ err = &ErrRpcStatus{
+ Code: int32(resp.Status.GetCode()),
+ Message: resp.GetStatus().GetMessage(),
+ }
}
sc.cli.doAfter(MessageHookPoints_ACK, messageCommons, duration,
messageHookPointsStatus)
- return nil
+ return err
}
func (sc *defaultSimpleConsumer) IsEndpointUpdated() bool {
diff --git a/golang/simple_consumer_test.go b/golang/simple_consumer_test.go
new file mode 100644
index 00000000..7021873e
--- /dev/null
+++ b/golang/simple_consumer_test.go
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package golang
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ v2 "github.com/apache/rocketmq-clients/golang/v5/protocol/v2"
+ "github.com/golang/mock/gomock"
+ "go.uber.org/zap"
+)
+
+func newSimpleConsumerForTest(t *testing.T) (*defaultSimpleConsumer,
*MockClientManager) {
+ t.Helper()
+ consumer, err := NewSimpleConsumer(&Config{
+ Endpoint: fakeAddress,
+ ConsumerGroup: MOCK_GROUP,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ sc := consumer.(*defaultSimpleConsumer)
+ manager := NewMockClientManager(gomock.NewController(t))
+ sc.cli.clientManager = manager
+ sc.cli.log = zap.NewNop().Sugar()
+ return sc, manager
+}
+
+// A non-OK status must surface as an error, otherwise the caller believes the
+// message was acknowledged while the broker still holds it invisible.
+func TestSimpleConsumerAckReportsNonOkStatus(t *testing.T) {
+ sc, manager := newSimpleConsumerForTest(t)
+
+ manager.EXPECT().AckMessage(gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.AckMessageRequest, time.Duration) (*v2.AckMessageResponse, error) {
+ return &v2.AckMessageResponse{
+ Status: &v2.Status{Code:
v2.Code_INVALID_RECEIPT_HANDLE, Message:
v2.Code_INVALID_RECEIPT_HANDLE.String()},
+ }, nil
+ }).Times(1)
+
+ err := sc.Ack(context.Background(), testMessageView())
+ var rpcErr *ErrRpcStatus
+ if !errors.As(err, &rpcErr) {
+ t.Fatalf("expected the non-OK ack status to surface as an
ErrRpcStatus, got %v", err)
+ }
+ if rpcErr.GetCode() != int32(v2.Code_INVALID_RECEIPT_HANDLE) {
+ t.Fatalf("expected code %d, got %d",
int32(v2.Code_INVALID_RECEIPT_HANDLE), rpcErr.GetCode())
+ }
+}
+
+// A failed change-invisible-duration request carries no new receipt handle;
+// adopting the empty one from the response would make every following call
fail.
+func TestSimpleConsumerChangeInvisibleDurationKeepsReceiptHandleOnError(t
*testing.T) {
+ sc, manager := newSimpleConsumerForTest(t)
+
+ manager.EXPECT().ChangeInvisibleDuration(gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, *v2.Endpoints,
*v2.ChangeInvisibleDurationRequest, time.Duration)
(*v2.ChangeInvisibleDurationResponse, error) {
+ return &v2.ChangeInvisibleDurationResponse{
+ Status: &v2.Status{Code:
v2.Code_INTERNAL_SERVER_ERROR, Message: "boom"},
+ ReceiptHandle: "",
+ }, nil
+ }).Times(1)
+
+ mv := testMessageView()
+ if err := sc.ChangeInvisibleDuration(mv, time.Second); err == nil {
+ t.Fatal("expected the non-OK status to surface as an error")
+ }
+ if mv.ReceiptHandle != "receipt-handle-0" {
+ t.Fatalf("a failed request must not overwrite the receipt
handle, got %q", mv.ReceiptHandle)
+ }
+}
