[Fix] Fix available permits in MessageReceived (#1181)

Fixes #1180

### Motivation
In the `MessageReceived`, the number of skipped messages should be increased to available permits to avoid skipped permits leading flow request not be sent.
---------

Co-authored-by: panjinjun <[email protected]>
(cherry picked from commit 5d25827)

pulsar/consumer_partition.go

@@ -1093,7 +1093,10 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	pc.metrics.MessagesReceived.Add(float64(numMsgs))
 	pc.metrics.PrefetchedMessages.Add(float64(numMsgs))
 
-	var bytesReceived int
+	var (
+		bytesReceived   int
+		skippedMessages int32
+	)
 	for i := 0; i < numMsgs; i++ {
 		smm, payload, err := reader.ReadMessage()
 		if err != nil || payload == nil {
@@ -1102,6 +1105,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		}
 		if ackSet != nil && !ackSet.Test(uint(i)) {
 			pc.log.Debugf("Ignoring message from %vth message, which has been acknowledged", i)
+			skippedMessages++
 			continue
 		}
 
@@ -1120,6 +1124,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 
 		if pc.messageShouldBeDiscarded(trackingMsgID) {
 			pc.AckID(trackingMsgID)
+			skippedMessages++
 			continue
 		}
 
@@ -1144,6 +1149,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		}
 
 		if pc.ackGroupingTracker.isDuplicate(msgID) {
+			skippedMessages++
 			continue
 		}
 
@@ -1218,6 +1224,10 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		pc.markScaleIfNeed()
 	}
 
+	if skippedMessages > 0 {
+		pc.availablePermits.add(skippedMessages)
+	}
+
 	// send messages to the dispatcher
 	pc.queueCh <- messages
 	return nil