Technoboy- commented on a change in pull request #13383:
URL: https://github.com/apache/pulsar/pull/13383#discussion_r779259574
##########
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:
Yes, if we enable `DeletionAtBatchIndexLevel`, this may exist the race
condition. One thread is for updating the cursorAckSet, another may get
relative value.
--
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]
