Technoboy- commented on a change in pull request #13383:
URL: https://github.com/apache/pulsar/pull/13383#discussion_r778581076
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
##########
@@ -476,6 +499,46 @@ public void doUnsubscribe(final long requestId) {
return completableFuture;
}
+ private long getBatchSize(MessageIdData msgId) {
+ long batchSize = 1;
+ if (Subscription.isIndividualAckMode(subType)) {
+ LongPair longPair = pendingAcks.get(msgId.getLedgerId(),
msgId.getEntryId());
+ // Consumer may ack the msg that not belongs to it.
+ if (longPair == null) {
+ Consumer ackOwnerConsumer =
getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
+ longPair =
ackOwnerConsumer.getPendingAcks().get(msgId.getLedgerId(), msgId.getEntryId());
+ if (longPair != null) {
+ batchSize = longPair.first;
+ }
+ } else {
+ batchSize = longPair.first;
+ }
+ }
+ return batchSize;
+ }
+
+ private long getAckedCount(PositionImpl position, long batchSize, long[]
ackSets) {
+ long ackedCount;
+ if (isDeletionAtBatchIndexLevelEnabled()) {
+ long[] cursorAckSet = getCursorAckSet(position);
+ if (cursorAckSet != null) {
+ BitSetRecyclable cursorBitSet =
BitSetRecyclable.create().resetWords(cursorAckSet);
+ int lastCardinality = cursorBitSet.cardinality();
+ BitSetRecyclable givenBitSet =
BitSetRecyclable.create().resetWords(ackSets);
+ cursorBitSet.and(givenBitSet);
+ givenBitSet.recycle();
+ int currentCardinality = cursorBitSet.cardinality();
+ ackedCount = lastCardinality - currentCardinality;
Review comment:
> Negative numbers may introduced here
Yes, it should be (batchSize - currentCardinality) - (batchSize -
lastCardinality) = the current ack count.
So lastCardinality - currentCardinality could not introduce negative.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
