when queue dropped,should'not consume the process queue.(#905)

Author: nick

consumer/pull_consumer.go

@@ -56,6 +56,14 @@ func (cr *ConsumeRequest) GetMsgList() []*primitive.MessageExt {
 	return cr.msgList
 }
 
+func (cr *ConsumeRequest) GetMQ() *primitive.MessageQueue {
+	return cr.messageQueue
+}
+
+func (cr *ConsumeRequest) GetPQ() *processQueue {
+	return cr.processQueue
+}
+
 type defaultPullConsumer struct {
 	*defaultConsumer
 
@@ -193,6 +201,14 @@ func (pc *defaultPullConsumer) Poll(ctx context.Context, timeout time.Duration)
 	case <-ctx.Done():
 		return nil, ErrNoNewMsg
 	case cr := <-pc.consumeRequestCache:
+		if cr.GetPQ().IsDroppd() {
+			rlog.Info("defaultPullConsumer poll the message queue not be able to consume, because it was dropped", map[string]interface{}{
+				rlog.LogKeyMessageQueue:  cr.GetMQ().String(),
+				rlog.LogKeyConsumerGroup: pc.consumerGroup,
+			})
+			return nil, ErrNoNewMsg
+		}
+
 		if len(cr.GetMsgList()) == 0 {
 			return nil, ErrNoNewMsg
 		}
@@ -780,6 +796,13 @@ func (pc *defaultPullConsumer) consumeMessageCurrently(pq *processQueue, mq *pri
 	if msgList == nil {
 		return
 	}
+	if pq.IsDroppd() {
+		rlog.Info("defaultPullConsumer consumeMessageCurrently the message queue not be able to consume, because it was dropped", map[string]interface{}{
+			rlog.LogKeyMessageQueue:  mq.String(),
+			rlog.LogKeyConsumerGroup: pc.consumerGroup,
+		})
+		return
+	}
 	cr := &ConsumeRequest{
 		messageQueue: mq,
 		processQueue: pq,