Fix bug consumer stop consume(issue-3131)
Motivation
Fix
When dispatcher is blocked and consumer blocked by un-ack messages, if consumer redelivery count < permitsReceivedWhileConsumerBlocked will cause dispatcher never send a message to the consumer again.
Modifications
Use permitsReceivedWhileConsumerBlocked to release dispatcher permits control.
Result
UT Passed.
@@ -654,12 +654,10 @@ public class Consumer {
subscription.redeliverUnacknowledgedMessages(this, pendingPositions);
msgRedeliver.recordMultipleEvents(totalRedeliveryMessages, totalRedeliveryMessages);
- int numberOfBlockedPermits = Math.min(totalRedeliveryMessages,
- PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.get(this));
+ int numberOfBlockedPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(this, 0);
// if permitsReceivedWhileConsumerBlocked has been accumulated then pass it to Dispatcher to flow messages
if (numberOfBlockedPermits > 0) {
- PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndAdd(this, -numberOfBlockedPermits);
MESSAGE_PERMITS_UPDATER.getAndAdd(this, numberOfBlockedPermits);
subscription.consumerFlow(this, numberOfBlockedPermits);
}
