Unable to batchreceive message the available permit is going to negative

the current version of pulsar client: 2.7.3
the available permit goes to negative .

consumer = pulsarClient.newConsumer()
.topic("persistent://xx/xxxx/xxx-1ms")
.ackTimeout(60, TimeUnit.SECONDS)
.subscriptionName("vmbTest")
// .subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
.batchReceivePolicy(BatchReceivePolicy.builder()
.maxNumMessages(100)
.maxNumBytes(1024 * 1024)
.timeout(1000, TimeUnit.MILLISECONDS)
.build()).subscriptionType(SubscriptionType.Shared)
.subscribe();

while (consumer.isConnected()) {
Messages batchMessages = consumer.batchReceive();
List messages = StreamSupport
.stream(batchMessages.spliterator(), false)
.map(msg -> String.valueOf(msg.getData()))
.collect(Collectors.toList());

consumer.acknowledge(batchMessages);
}
consumer.close();

} catch (PulsarClientException e) {
if (consumer != null) {
consumer.close();
}
}

When does the available permit increment again

Start with the Java consumer batchReceive() entry point and the BatchReceivePolicy configuration shown in the report. Trace how available permits are consumed and replenished while reproducing the example with Pulsar client 2.7.3. Done means the permit behavior is explained and the reported negative value is either corrected or confirmed as expected.

Written by the indexing model from the issue text.