Add consumer state check when request commands (apache#772)

* Add consumer state check when request commands

Signed-off-by: xiaolongran <xiaolongran@tencent.com>

* fix a little

Signed-off-by: xiaolongran <xiaolongran@tencent.com>

Author: wolfstudy

pulsar/consumer_partition.go

@@ -18,6 +18,7 @@
 package pulsar
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"strings"
@@ -278,6 +279,10 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 }
 
 func (pc *partitionConsumer) getLastMessageID() (trackingMessageID, error) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
+		return trackingMessageID{}, errors.New("failed to redeliver closing or closed consumer")
+	}
 	req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
 	pc.eventsCh <- req
 
@@ -292,6 +297,11 @@ func (pc *partitionConsumer) internalGetLastMessageID(req *getLastMsgIDRequest)
 }
 
 func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to getLastMessageID closing or closed consumer")
+		return trackingMessageID{}, errors.New("failed to getLastMessageID closing or closed consumer")
+	}
+
 	requestID := pc.client.rpcClient.NewRequestID()
 	cmdGetLastMessageID := &pb.CommandGetLastMessageId{
 		RequestId:  proto.Uint64(requestID),
@@ -308,6 +318,10 @@ func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error
 }
 
 func (pc *partitionConsumer) AckID(msgID trackingMessageID) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return
+	}
 	if !msgID.Undefined() && msgID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-msgID.receivedTime.UnixNano()) / 1.0e9)
@@ -331,6 +345,10 @@ func (pc *partitionConsumer) NackMsg(msg Message) {
 }
 
 func (pc *partitionConsumer) Redeliver(msgIds []messageID) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
+		return
+	}
 	pc.eventsCh <- &redeliveryRequest{msgIds}
 
 	iMsgIds := make([]MessageID, len(msgIds))
@@ -341,6 +359,10 @@ func (pc *partitionConsumer) Redeliver(msgIds []messageID) {
 }
 
 func (pc *partitionConsumer) internalRedeliver(req *redeliveryRequest) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
+		return
+	}
 	msgIds := req.msgIds
 	pc.log.Debug("Request redelivery after negative ack for messages", msgIds)
 
@@ -352,11 +374,14 @@ func (pc *partitionConsumer) internalRedeliver(req *redeliveryRequest) {
 		}
 	}
 
-	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(),
+	err := pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(),
 		pb.BaseCommand_REDELIVER_UNACKNOWLEDGED_MESSAGES, &pb.CommandRedeliverUnacknowledgedMessages{
 			ConsumerId: proto.Uint64(pc.consumerID),
 			MessageIds: msgIDDataList,
 		})
+	if err != nil {
+		pc.log.Error("Connection was closed when request redeliver cmd")
+	}
 }
 
 func (pc *partitionConsumer) getConsumerState() consumerState {
@@ -381,6 +406,10 @@ func (pc *partitionConsumer) Close() {
 }
 
 func (pc *partitionConsumer) Seek(msgID trackingMessageID) error {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to seek by closing or closed consumer")
+		return errors.New("failed to seek by closing or closed consumer")
+	}
 	req := &seekRequest{
 		doneCh: make(chan struct{}),
 		msgID:  msgID,
@@ -407,7 +436,7 @@ func (pc *partitionConsumer) requestSeek(msgID messageID) error {
 func (pc *partitionConsumer) requestSeekWithoutClear(msgID messageID) error {
 	state := pc.getConsumerState()
 	if state == consumerClosing || state == consumerClosed {
-		pc.log.WithField("state", state).Error("Consumer is closing or has closed")
+		pc.log.WithField("state", state).Error("failed seek by consumer is closing or has closed")
 		return nil
 	}
 
@@ -434,6 +463,10 @@ func (pc *partitionConsumer) requestSeekWithoutClear(msgID messageID) error {
 }
 
 func (pc *partitionConsumer) SeekByTime(time time.Time) error {
+	if state := pc.getConsumerState(); state == consumerClosing || state == consumerClosed {
+		pc.log.WithField("state", pc.state).Error("Failed seekByTime by consumer is closing or has closed")
+		return errors.New("failed seekByTime by consumer is closing or has closed")
+	}
 	req := &seekByTimeRequest{
 		doneCh:      make(chan struct{}),
 		publishTime: time,
@@ -450,7 +483,7 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 
 	state := pc.getConsumerState()
 	if state == consumerClosing || state == consumerClosed {
-		pc.log.WithField("state", pc.state).Error("Consumer is closing or has closed")
+		pc.log.WithField("state", pc.state).Error("Failed seekByTime by consumer is closing or has closed")
 		return
 	}
 
@@ -477,6 +510,10 @@ func (pc *partitionConsumer) clearMessageChannels() {
 }
 
 func (pc *partitionConsumer) internalAck(req *ackRequest) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return
+	}
 	msgID := req.msgID
 
 	messageIDs := make([]*pb.MessageIdData, 1)
@@ -491,7 +528,10 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 		AckType:    pb.CommandAck_Individual.Enum(),
 	}
 
-	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_ACK, cmdAck)
+	err := pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_ACK, cmdAck)
+	if err != nil {
+		pc.log.Error("Connection was closed when request ack cmd")
+	}
 }
 
 func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, headersAndPayload internal.Buffer) error {
@@ -728,6 +768,10 @@ func (pc *partitionConsumer) ConnectionClosed() {
 // before the application is ready to consume them. After the consumer is ready,
 // the client needs to give permission to the broker to push messages.
 func (pc *partitionConsumer) internalFlow(permits uint32) error {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
+		return errors.New("consumer closing or closed")
+	}
 	if permits == 0 {
 		return fmt.Errorf("invalid number of permits requested: %d", permits)
 	}
@@ -736,7 +780,11 @@ func (pc *partitionConsumer) internalFlow(permits uint32) error {
 		ConsumerId:     proto.Uint64(pc.consumerID),
 		MessagePermits: proto.Uint32(permits),
 	}
-	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_FLOW, cmdFlow)
+	err := pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(), pb.BaseCommand_FLOW, cmdFlow)
+	if err != nil {
+		pc.log.Error("Connection was closed when request flow cmd")
+		return err
+	}
 
 	return nil
 }
@@ -1259,18 +1307,26 @@ func (pc *partitionConsumer) initializeCompressionProvider(
 
 func (pc *partitionConsumer) discardCorruptedMessage(msgID *pb.MessageIdData,
 	validationError pb.CommandAck_ValidationError) {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to discardCorruptedMessage " +
+			"by closing or closed consumer")
+		return
+	}
 	pc.log.WithFields(log.Fields{
 		"msgID":           msgID,
 		"validationError": validationError,
 	}).Error("Discarding corrupted message")
 
-	pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(),
+	err := pc.client.rpcClient.RequestOnCnxNoWait(pc._getConn(),
 		pb.BaseCommand_ACK, &pb.CommandAck{
 			ConsumerId:      proto.Uint64(pc.consumerID),
 			MessageId:       []*pb.MessageIdData{msgID},
 			AckType:         pb.CommandAck_Individual.Enum(),
 			ValidationError: validationError.Enum(),
 		})
+	if err != nil {
+		pc.log.Error("Connection was closed when request ack cmd")
+	}
 }
 
 // _setConn sets the internal connection field of this partition consumer atomically.