codelipenghui commented on a change in pull request #13383:
URL: https://github.com/apache/pulsar/pull/13383#discussion_r776683141
##########
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);
Review comment:
Looks like maybe encounter a race condition here, the cursor ack set
might be changed by another thread.
##########
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
--
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]
