fix: queue to exit when mq processing msg panic (#1057)

* fix queue to exit when mq processing msg panic --------- Co-authored-by: 桂千山 <[email protected]>
douyu · Jan 26, 2024 · 07fa9b2 · 07fa9b2
1 parent 299b3ec
commit 07fa9b2
Showing 1 changed file with 20 additions and 7 deletions.
diff --git a/pkg/client/rocketmq/push_consumer.go b/pkg/client/rocketmq/push_consumer.go
@@ -16,6 +16,8 @@ package rocketmq
 
 import (
 	"context"
+	"errors"
+	"runtime/debug"
 
 	"github.com/apache/rocketmq-client-go/v2"
 	"github.com/apache/rocketmq-client-go/v2/consumer"
@@ -102,7 +104,14 @@ func (cc *PushConsumer) RegisterSingleMessage(f func(context.Context, *primitive
 		semconv.MessagingRocketmqConsumptionModelKey.String(cc.MessageModel),
 	}
 
-	fn := func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
+	fn := func(ctx context.Context, msgs ...*primitive.MessageExt) (result consumer.ConsumeResult, err error) {
+		// the recover to prevent panic from causing the coroutine to exit when processing msg.
+		defer func() {
+			if r := recover(); r != nil {
+				xlog.Jupiter().Error("consumer message panic", zap.String("stack", string(debug.Stack())))
+				result, err = consumer.ConsumeRetryLater, errors.New("consumer message panic")
+			}
+		}()
 		for _, msg := range msgs {
 			var (
 				span trace.Span
@@ -135,8 +144,7 @@ func (cc *PushConsumer) RegisterSingleMessage(f func(context.Context, *primitive
 				}
 			}
 
-			err := f(ctx, msg)
-			if err != nil {
+			if err := f(ctx, msg); err != nil {
 				xlog.Jupiter().Error("consumer message", zap.Error(err), zap.String("field", cc.name), zap.Any("ext", msg))
 				if cc.EnableTrace && span != nil {
 					span.RecordError(err)
@@ -158,8 +166,14 @@ func (cc *PushConsumer) RegisterBatchMessage(f func(context.Context, ...*primiti
 
 	tracer := xtrace.NewTracer(trace.SpanKindConsumer)
 
-	fn := func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
-
+	fn := func(ctx context.Context, msgs ...*primitive.MessageExt) (result consumer.ConsumeResult, err error) {
+		// the recover to prevent panic from causing the coroutine to exit when processing msg.
+		defer func() {
+			if r := recover(); r != nil {
+				xlog.Jupiter().Error("consumer message panic", zap.String("stack", string(debug.Stack())))
+				result, err = consumer.ConsumeRetryLater, errors.New("consumer message panic")
+			}
+		}()
 		if cc.EnableTrace {
 			for _, msg := range msgs {
 				var (
@@ -186,8 +200,7 @@ func (cc *PushConsumer) RegisterBatchMessage(f func(context.Context, ...*primiti
 			}
 		}
 
-		err := f(ctx, msgs...)
-		if err != nil {
+		if err := f(ctx, msgs...); err != nil {
 			xlog.Jupiter().Error("consumer batch message", zap.Error(err), zap.String("field", cc.name))
 			return consumer.ConsumeRetryLater, err
 		}