congbobo184 commented on code in PR #19031:
URL: https://github.com/apache/pulsar/pull/19031#discussion_r1056136563
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -573,7 +644,26 @@ protected CompletableFuture<Void>
doAcknowledge(List<MessageId> messageIdList, A
return doTransactionAcknowledgeForResponse(messageIdList, ackType,
null,
properties, new TxnID(txn.getTxnIdMostBits(),
txn.getTxnIdLeastBits()));
} else {
- return
this.acknowledgmentsGroupingTracker.addListAcknowledgment(messageIdList,
ackType, properties);
+ for (MessageId messageId : messageIdList) {
+ checkArgument(messageId instanceof MessageIdImpl);
Review Comment:
has checked in line-631, may can delete line-631 checked
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -549,13 +599,34 @@ 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);
+ if (messageIdImpl == null) {
+ return CompletableFuture.completedFuture(null);
+ } else if (messageIdImpl instanceof BatchMessageIdImpl) {
+ return acknowledgmentsGroupingTracker.addBatchIndexAck(
+ (BatchMessageIdImpl) messageIdImpl, ackType,
properties);
+ } else {
+ processMessageIdBeforeAcknowledge(messageIdImpl, ackType,
batchMessageId.getOriginalBatchSize());
+ return
acknowledgmentsGroupingTracker.addAcknowledgment(messageIdImpl, ackType,
properties);
+ }
+ } else {
+ MessageIdImpl messageIdImpl = (MessageIdImpl) messageId;
+ processMessageIdBeforeAcknowledge(messageIdImpl, ackType, 1);
+ return
acknowledgmentsGroupingTracker.addAcknowledgment(messageIdImpl, ackType,
properties);
+ }
}
@Override
protected CompletableFuture<Void> doAcknowledge(List<MessageId>
messageIdList, AckType ackType,
Map<String, Long>
properties, TransactionImpl txn) {
-
+ List<MessageIdImpl> messageIdListToAck = new ArrayList<>();
Review Comment:
move to `checkArgument(messageId instanceof MessageIdImpl);` below
--
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]
