BewareMyPower commented on code in PR #19031:
URL: https://github.com/apache/pulsar/pull/19031#discussion_r1058725134
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -549,13 +600,33 @@ protected CompletableFuture<Void> doAcknowledge(MessageId
messageId, AckType ack
return doTransactionAcknowledgeForResponse(messageId, ackType,
null, properties,
new TxnID(txn.getTxnIdMostBits(),
txn.getTxnIdLeastBits()));
}
- return
acknowledgmentsGroupingTracker.addAcknowledgment((MessageIdImpl) messageId,
ackType, properties);
+ if (ackType == AckType.Individual) {
+ onAcknowledge(messageId, null);
+ } else {
+ onAcknowledgeCumulative(messageId, null);
+ }
+ if (messageId instanceof BatchMessageIdImpl) {
+ BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
+ MessageIdImpl messageIdImpl =
getMessageIdToAcknowledge(batchMessageId, ackType);
Review Comment:
No. This method is to get the `MessageIdImpl` to acknowledge.
`increaseAndGet` modifies the atomic variable itself, but
`getMessageIdToAcknowledge` just calculated a `MessageIdImpl` to acknowledge
from the `BatchMessageIdImpl` and does not modify the `BatchMessageIdImpl`
itself.
--
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]
