Skip to content

Commit f5eebae

Browse files
shibdsrinath-ctds
shibd
authored and
srinath-ctds
committed
[improve][broker] Optimize the performance of individual acknowledgments (apache#23072)
(cherry picked from commit 77b6378) (cherry picked from commit 487913e)
1 parent 148077c commit f5eebae

File tree

1 file changed

+69
-82
lines changed
  • pulsar-broker/src/main/java/org/apache/pulsar/broker/service

1 file changed

+69
-82
lines changed

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

+69-82
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.bookkeeper.mledger.impl.PositionImpl;
4444
import org.apache.commons.lang3.mutable.MutableInt;
4545
import org.apache.commons.lang3.tuple.MutablePair;
46+
import org.apache.commons.lang3.tuple.Pair;
4647
import org.apache.pulsar.broker.ServiceConfiguration;
4748
import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription;
4849
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
@@ -506,14 +507,16 @@ public CompletableFuture<Void> messageAcked(CommandAck ack) {
506507

507508
//this method is for individual ack not carry the transaction
508509
private CompletableFuture<Long> individualAckNormal(CommandAck ack, Map<String, Long> properties) {
509-
List<Position> positionsAcked = new ArrayList<>();
510+
List<Pair<Consumer, Position>> positionsAcked = new ArrayList<>();
510511
long totalAckCount = 0;
511512
for (int i = 0; i < ack.getMessageIdsCount(); i++) {
512513
MessageIdData msgId = ack.getMessageIdAt(i);
513514
PositionImpl position;
514-
long ackedCount = 0;
515-
long batchSize = getBatchSize(msgId);
516-
Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
515+
Pair<Consumer, Long> ackOwnerConsumerAndBatchSize =
516+
getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(), msgId.getEntryId());
517+
Consumer ackOwnerConsumer = ackOwnerConsumerAndBatchSize.getLeft();
518+
long ackedCount;
519+
long batchSize = ackOwnerConsumerAndBatchSize.getRight();
517520
if (msgId.getAckSetsCount() > 0) {
518521
long[] ackSets = new long[msgId.getAckSetsCount()];
519522
for (int j = 0; j < msgId.getAckSetsCount(); j++) {
@@ -532,28 +535,32 @@ private CompletableFuture<Long> individualAckNormal(CommandAck ack, Map<String,
532535
} else {
533536
position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId());
534537
ackedCount = getAckedCountForMsgIdNoAckSets(batchSize, position, ackOwnerConsumer);
535-
if (checkCanRemovePendingAcksAndHandle(position, msgId)) {
538+
if (checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId)) {
536539
addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
537540
}
538541
}
539542

540-
positionsAcked.add(position);
543+
positionsAcked.add(Pair.of(ackOwnerConsumer, position));
541544

542545
checkAckValidationError(ack, position);
543546

544547
totalAckCount += ackedCount;
545548
}
546-
subscription.acknowledgeMessage(positionsAcked, AckType.Individual, properties);
549+
subscription.acknowledgeMessage(positionsAcked.stream()
550+
.map(Pair::getRight)
551+
.collect(Collectors.toList()), AckType.Individual, properties);
547552
CompletableFuture<Long> completableFuture = new CompletableFuture<>();
548553
completableFuture.complete(totalAckCount);
549554
if (isTransactionEnabled() && Subscription.isIndividualAckMode(subType)) {
550-
completableFuture.whenComplete((v, e) -> positionsAcked.forEach(position -> {
555+
completableFuture.whenComplete((v, e) -> positionsAcked.forEach(positionPair -> {
556+
Consumer ackOwnerConsumer = positionPair.getLeft();
557+
Position position = positionPair.getRight();
551558
//check if the position can remove from the consumer pending acks.
552559
// the bit set is empty in pending ack handle.
553560
if (((PositionImpl) position).getAckSet() != null) {
554561
if (((PersistentSubscription) subscription)
555562
.checkIsCanDeleteConsumerPendingAck((PositionImpl) position)) {
556-
removePendingAcks((PositionImpl) position);
563+
removePendingAcks(ackOwnerConsumer, (PositionImpl) position);
557564
}
558565
}
559566
}));
@@ -565,7 +572,7 @@ private CompletableFuture<Long> individualAckNormal(CommandAck ack, Map<String,
565572
//this method is for individual ack carry the transaction
566573
private CompletableFuture<Long> individualAckWithTransaction(CommandAck ack) {
567574
// Individual ack
568-
List<MutablePair<PositionImpl, Integer>> positionsAcked = new ArrayList<>();
575+
List<Pair<Consumer, MutablePair<PositionImpl, Integer>>> positionsAcked = new ArrayList<>();
569576
if (!isTransactionEnabled()) {
570577
return FutureUtil.failedFuture(
571578
new BrokerServiceException.NotAllowedException("Server don't support transaction ack!"));
@@ -575,20 +582,23 @@ private CompletableFuture<Long> individualAckWithTransaction(CommandAck ack) {
575582
for (int i = 0; i < ack.getMessageIdsCount(); i++) {
576583
MessageIdData msgId = ack.getMessageIdAt(i);
577584
PositionImpl position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId());
585+
Consumer ackOwnerConsumer = getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(),
586+
msgId.getEntryId()).getLeft();
578587
// acked count at least one
579-
long ackedCount = 0;
580-
long batchSize = 0;
588+
long ackedCount;
589+
long batchSize;
581590
if (msgId.hasBatchSize()) {
582591
batchSize = msgId.getBatchSize();
583592
// ack batch messages set ackeCount = batchSize
584593
ackedCount = msgId.getBatchSize();
585-
positionsAcked.add(new MutablePair<>(position, msgId.getBatchSize()));
594+
positionsAcked.add(Pair.of(ackOwnerConsumer, new MutablePair<>(position, msgId.getBatchSize())));
586595
} else {
587596
// ack no batch message set ackedCount = 1
597+
batchSize = 0;
588598
ackedCount = 1;
589-
positionsAcked.add(new MutablePair<>(position, (int) batchSize));
599+
positionsAcked.add(Pair.of(ackOwnerConsumer, new MutablePair<>(position, (int) batchSize)));
590600
}
591-
Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
601+
592602
if (msgId.getAckSetsCount() > 0) {
593603
long[] ackSets = new long[msgId.getAckSetsCount()];
594604
for (int j = 0; j < msgId.getAckSetsCount(); j++) {
@@ -600,47 +610,31 @@ private CompletableFuture<Long> individualAckWithTransaction(CommandAck ack) {
600610

601611
addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
602612

603-
checkCanRemovePendingAcksAndHandle(position, msgId);
613+
checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId);
604614

605615
checkAckValidationError(ack, position);
606616

607617
totalAckCount.add(ackedCount);
608618
}
609619

610620
CompletableFuture<Void> completableFuture = transactionIndividualAcknowledge(ack.getTxnidMostBits(),
611-
ack.getTxnidLeastBits(), positionsAcked);
621+
ack.getTxnidLeastBits(), positionsAcked.stream().map(Pair::getRight).collect(Collectors.toList()));
612622
if (Subscription.isIndividualAckMode(subType)) {
613623
completableFuture.whenComplete((v, e) ->
614-
positionsAcked.forEach(positionLongMutablePair -> {
624+
positionsAcked.forEach(positionPair -> {
625+
Consumer ackOwnerConsumer = positionPair.getLeft();
626+
MutablePair<PositionImpl, Integer> positionLongMutablePair = positionPair.getRight();
615627
if (positionLongMutablePair.getLeft().getAckSet() != null) {
616628
if (((PersistentSubscription) subscription)
617629
.checkIsCanDeleteConsumerPendingAck(positionLongMutablePair.left)) {
618-
removePendingAcks(positionLongMutablePair.left);
630+
removePendingAcks(ackOwnerConsumer, positionLongMutablePair.left);
619631
}
620632
}
621633
}));
622634
}
623635
return completableFuture.thenApply(__ -> totalAckCount.sum());
624636
}
625637

626-
private long getBatchSize(MessageIdData msgId) {
627-
long batchSize = 1;
628-
if (Subscription.isIndividualAckMode(subType)) {
629-
LongPair longPair = pendingAcks.get(msgId.getLedgerId(), msgId.getEntryId());
630-
// Consumer may ack the msg that not belongs to it.
631-
if (longPair == null) {
632-
Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
633-
longPair = ackOwnerConsumer.getPendingAcks().get(msgId.getLedgerId(), msgId.getEntryId());
634-
if (longPair != null) {
635-
batchSize = longPair.first;
636-
}
637-
} else {
638-
batchSize = longPair.first;
639-
}
640-
}
641-
return batchSize;
642-
}
643-
644638
private long getAckedCountForMsgIdNoAckSets(long batchSize, PositionImpl position, Consumer consumer) {
645639
if (isAcknowledgmentAtBatchIndexLevelEnabled && Subscription.isIndividualAckMode(subType)) {
646640
long[] cursorAckSet = getCursorAckSet(position);
@@ -700,26 +694,39 @@ private void checkAckValidationError(CommandAck ack, PositionImpl position) {
700694
}
701695
}
702696

703-
private boolean checkCanRemovePendingAcksAndHandle(PositionImpl position, MessageIdData msgId) {
697+
private boolean checkCanRemovePendingAcksAndHandle(Consumer ackOwnedConsumer,
698+
PositionImpl position, MessageIdData msgId) {
704699
if (Subscription.isIndividualAckMode(subType) && msgId.getAckSetsCount() == 0) {
705-
return removePendingAcks(position);
700+
return removePendingAcks(ackOwnedConsumer, position);
706701
}
707702
return false;
708703
}
709704

710-
private Consumer getAckOwnerConsumer(long ledgerId, long entryId) {
711-
Consumer ackOwnerConsumer = this;
705+
/**
706+
* Retrieves the acknowledgment owner consumer and batch size for the specified ledgerId and entryId.
707+
*
708+
* @param ledgerId The ID of the ledger.
709+
* @param entryId The ID of the entry.
710+
* @return Pair<Consumer, BatchSize>
711+
*/
712+
private Pair<Consumer, Long> getAckOwnerConsumerAndBatchSize(long ledgerId, long entryId) {
712713
if (Subscription.isIndividualAckMode(subType)) {
713-
if (!getPendingAcks().containsKey(ledgerId, entryId)) {
714+
LongPair longPair = getPendingAcks().get(ledgerId, entryId);
715+
if (longPair != null) {
716+
return Pair.of(this, longPair.first);
717+
} else {
718+
// If there are more consumers, this step will consume more CPU, and it should be optimized later.
714719
for (Consumer consumer : subscription.getConsumers()) {
715-
if (consumer != this && consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
716-
ackOwnerConsumer = consumer;
717-
break;
720+
if (consumer != this) {
721+
longPair = consumer.getPendingAcks().get(ledgerId, entryId);
722+
if (longPair != null) {
723+
return Pair.of(consumer, longPair.first);
724+
}
718725
}
719726
}
720727
}
721728
}
722-
return ackOwnerConsumer;
729+
return Pair.of(this, 1L);
723730
}
724731

725732
private long[] getCursorAckSet(PositionImpl position) {
@@ -971,44 +978,24 @@ public int hashCode() {
971978
*
972979
* @param position
973980
*/
974-
private boolean removePendingAcks(PositionImpl position) {
975-
Consumer ackOwnedConsumer = null;
976-
if (pendingAcks.get(position.getLedgerId(), position.getEntryId()) == null) {
977-
for (Consumer consumer : subscription.getConsumers()) {
978-
if (!consumer.equals(this) && consumer.getPendingAcks().containsKey(position.getLedgerId(),
979-
position.getEntryId())) {
980-
ackOwnedConsumer = consumer;
981-
break;
982-
}
983-
}
984-
} else {
985-
ackOwnedConsumer = this;
981+
private boolean removePendingAcks(Consumer ackOwnedConsumer, PositionImpl position) {
982+
if (!ackOwnedConsumer.getPendingAcks().remove(position.getLedgerId(), position.getEntryId())) {
983+
// Message was already removed by the other consumer
984+
return false;
986985
}
987-
988-
// remove pending message from appropriate consumer and unblock unAckMsg-flow if requires
989-
LongPair ackedPosition = ackOwnedConsumer != null
990-
? ackOwnedConsumer.getPendingAcks().get(position.getLedgerId(), position.getEntryId())
991-
: null;
992-
if (ackedPosition != null) {
993-
if (!ackOwnedConsumer.getPendingAcks().remove(position.getLedgerId(), position.getEntryId())) {
994-
// Message was already removed by the other consumer
995-
return false;
996-
}
997-
if (log.isDebugEnabled()) {
998-
log.debug("[{}-{}] consumer {} received ack {}", topicName, subscription, consumerId, position);
999-
}
1000-
// unblock consumer-throttling when limit check is disabled or receives half of maxUnackedMessages =>
1001-
// consumer can start again consuming messages
1002-
int unAckedMsgs = UNACKED_MESSAGES_UPDATER.get(ackOwnedConsumer);
1003-
if ((((unAckedMsgs <= getMaxUnackedMessages() / 2) && ackOwnedConsumer.blockedConsumerOnUnackedMsgs)
1004-
&& ackOwnedConsumer.shouldBlockConsumerOnUnackMsgs())
1005-
|| !shouldBlockConsumerOnUnackMsgs()) {
1006-
ackOwnedConsumer.blockedConsumerOnUnackedMsgs = false;
1007-
flowConsumerBlockedPermits(ackOwnedConsumer);
1008-
}
1009-
return true;
986+
if (log.isDebugEnabled()) {
987+
log.debug("[{}-{}] consumer {} received ack {}", topicName, subscription, consumerId, position);
1010988
}
1011-
return false;
989+
// unblock consumer-throttling when limit check is disabled or receives half of maxUnackedMessages =>
990+
// consumer can start again consuming messages
991+
int unAckedMsgs = UNACKED_MESSAGES_UPDATER.get(ackOwnedConsumer);
992+
if ((((unAckedMsgs <= getMaxUnackedMessages() / 2) && ackOwnedConsumer.blockedConsumerOnUnackedMsgs)
993+
&& ackOwnedConsumer.shouldBlockConsumerOnUnackMsgs())
994+
|| !shouldBlockConsumerOnUnackMsgs()) {
995+
ackOwnedConsumer.blockedConsumerOnUnackedMsgs = false;
996+
flowConsumerBlockedPermits(ackOwnedConsumer);
997+
}
998+
return true;
1012999
}
10131000

10141001
public ConcurrentLongLongPairHashMap getPendingAcks() {

0 commit comments

Comments
 (0)