Ignore the send on closed channel error

Author: BewareMyPower

pulsar/consumer_partition.go

@@ -1138,6 +1138,15 @@ func (pc *partitionConsumer) internalAckList(request *ackListRequest) {
 }
 
 func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, headersAndPayload internal.Buffer) error {
+	defer func() {
+		if r := recover(); r != nil {
+			if err, ok := r.(error); ok && err.Error() == "send on closed channel" {
+				pc.log.WithField("panic", r).Debug("panic recovered in MessageReceived")
+				return
+			}
+			panic(r)
+		}
+	}()
 	pbMsgID := response.GetMessageId()
 
 	reader := internal.NewMessageReader(headersAndPayload)