From 618e710f61e03dfb3431df87f159220dde749171 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Wed, 24 Jul 2024 20:12:18 +0800 Subject: [PATCH 1/2] [improve][broker] Optimize the performance of individual acknowledgments --- .../pulsar/broker/service/Consumer.java | 150 ++++++++---------- 1 file changed, 68 insertions(+), 82 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index dca64395d8674..8af98bbfd45cd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -47,6 +47,7 @@ import org.apache.bookkeeper.mledger.impl.AckSetStateUtil; import org.apache.commons.lang3.mutable.MutableInt; import org.apache.commons.lang3.tuple.MutablePair; +import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription; import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; @@ -531,14 +532,15 @@ public CompletableFuture messageAcked(CommandAck ack) { //this method is for individual ack not carry the transaction private CompletableFuture individualAckNormal(CommandAck ack, Map properties) { - List positionsAcked = new ArrayList<>(); + List> positionsAcked = new ArrayList<>(); long totalAckCount = 0; for (int i = 0; i < ack.getMessageIdsCount(); i++) { MessageIdData msgId = ack.getMessageIdAt(i); Position position; - long ackedCount = 0; - long batchSize = getBatchSize(msgId); - Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); + Pair ackOwnerConsumerPair = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); + Consumer ackOwnerConsumer = ackOwnerConsumerPair.getLeft(); + long ackedCount; + long batchSize = ackOwnerConsumerPair.getRight(); if (msgId.getAckSetsCount() > 0) { long[] ackSets = new long[msgId.getAckSetsCount()]; for (int j = 0; j < msgId.getAckSetsCount(); j++) { @@ -557,28 +559,32 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map completableFuture = new CompletableFuture<>(); completableFuture.complete(totalAckCount); if (isTransactionEnabled() && Subscription.isIndividualAckMode(subType)) { - completableFuture.whenComplete((v, e) -> positionsAcked.forEach(position -> { + completableFuture.whenComplete((v, e) -> positionsAcked.forEach(positionPair -> { + Consumer ackOwnerConsumer = positionPair.getLeft(); + Position position = positionPair.getRight(); //check if the position can remove from the consumer pending acks. // the bit set is empty in pending ack handle. if (AckSetStateUtil.hasAckSet(position)) { if (((PersistentSubscription) subscription) .checkIsCanDeleteConsumerPendingAck(position)) { - removePendingAcks(position); + removePendingAcks(ackOwnerConsumer, position); } } })); @@ -590,7 +596,7 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map individualAckWithTransaction(CommandAck ack) { // Individual ack - List> positionsAcked = new ArrayList<>(); + List>> positionsAcked = new ArrayList<>(); if (!isTransactionEnabled()) { return FutureUtil.failedFuture( new BrokerServiceException.NotAllowedException("Server don't support transaction ack!")); @@ -600,20 +606,23 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { for (int i = 0; i < ack.getMessageIdsCount(); i++) { MessageIdData msgId = ack.getMessageIdAt(i); Position position = AckSetStateUtil.createPositionWithAckSet(msgId.getLedgerId(), msgId.getEntryId(), null); + Pair ackOwnerConsumerPair = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); + Consumer ackOwnerConsumer = ackOwnerConsumerPair.getLeft(); // acked count at least one - long ackedCount = 0; - long batchSize = 0; + long ackedCount; + long batchSize; if (msgId.hasBatchSize()) { batchSize = msgId.getBatchSize(); // ack batch messages set ackeCount = batchSize ackedCount = msgId.getBatchSize(); - positionsAcked.add(new MutablePair<>(position, msgId.getBatchSize())); + positionsAcked.add(Pair.of(ackOwnerConsumer, new MutablePair<>(position, msgId.getBatchSize()))); } else { // ack no batch message set ackedCount = 1 + batchSize = 0; ackedCount = 1; - positionsAcked.add(new MutablePair<>(position, (int) batchSize)); + positionsAcked.add(Pair.of(ackOwnerConsumer, new MutablePair<>(position, (int) batchSize))); } - Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); + if (msgId.getAckSetsCount() > 0) { long[] ackSets = new long[msgId.getAckSetsCount()]; for (int j = 0; j < msgId.getAckSetsCount(); j++) { @@ -625,7 +634,7 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); - checkCanRemovePendingAcksAndHandle(position, msgId); + checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId); checkAckValidationError(ack, position); @@ -633,14 +642,16 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { } CompletableFuture completableFuture = transactionIndividualAcknowledge(ack.getTxnidMostBits(), - ack.getTxnidLeastBits(), positionsAcked); + ack.getTxnidLeastBits(), positionsAcked.stream().map(Pair::getRight).collect(Collectors.toList())); if (Subscription.isIndividualAckMode(subType)) { completableFuture.whenComplete((v, e) -> - positionsAcked.forEach(positionLongMutablePair -> { + positionsAcked.forEach(positionPair -> { + Consumer ackOwnerConsumer = positionPair.getLeft(); + MutablePair positionLongMutablePair = positionPair.getRight(); if (AckSetStateUtil.hasAckSet(positionLongMutablePair.getLeft())) { if (((PersistentSubscription) subscription) .checkIsCanDeleteConsumerPendingAck(positionLongMutablePair.left)) { - removePendingAcks(positionLongMutablePair.left); + removePendingAcks(ackOwnerConsumer, positionLongMutablePair.left); } } })); @@ -648,24 +659,6 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { return completableFuture.thenApply(__ -> totalAckCount.sum()); } - 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 getAckedCountForMsgIdNoAckSets(long batchSize, Position position, Consumer consumer) { if (isAcknowledgmentAtBatchIndexLevelEnabled && Subscription.isIndividualAckMode(subType)) { long[] cursorAckSet = getCursorAckSet(position); @@ -725,26 +718,39 @@ private void checkAckValidationError(CommandAck ack, Position position) { } } - private boolean checkCanRemovePendingAcksAndHandle(Position position, MessageIdData msgId) { + private boolean checkCanRemovePendingAcksAndHandle(Consumer ackOwnedConsumer, + Position position, MessageIdData msgId) { if (Subscription.isIndividualAckMode(subType) && msgId.getAckSetsCount() == 0) { - return removePendingAcks(position); + return removePendingAcks(ackOwnedConsumer, position); } return false; } - private Consumer getAckOwnerConsumer(long ledgerId, long entryId) { - Consumer ackOwnerConsumer = this; + /** + * Retrieves the acknowledgment owner consumer and batch size for the specified ledgerId and entryId. + * + * @param ledgerId The ID of the ledger. + * @param entryId The ID of the entry. + * @return Pair + */ + private Pair getAckOwnerConsumer(long ledgerId, long entryId) { if (Subscription.isIndividualAckMode(subType)) { - if (!getPendingAcks().containsKey(ledgerId, entryId)) { + LongPair longPair = getPendingAcks().get(ledgerId, entryId); + if (longPair != null) { + return Pair.of(this, longPair.first); + } else { + // If there are more consumers, this step will consume more CPU, and it should be optimized later. for (Consumer consumer : subscription.getConsumers()) { - if (consumer != this && consumer.getPendingAcks().containsKey(ledgerId, entryId)) { - ackOwnerConsumer = consumer; - break; + if (consumer != this) { + longPair = consumer.getPendingAcks().get(ledgerId, entryId); + if (longPair != null) { + return Pair.of(consumer, longPair.first); + } } } } } - return ackOwnerConsumer; + return Pair.of(this, 1L); } private long[] getCursorAckSet(Position position) { @@ -1019,44 +1025,24 @@ public int hashCode() { * * @param position */ - private boolean removePendingAcks(Position position) { - Consumer ackOwnedConsumer = null; - if (pendingAcks.get(position.getLedgerId(), position.getEntryId()) == null) { - for (Consumer consumer : subscription.getConsumers()) { - if (!consumer.equals(this) && consumer.getPendingAcks().containsKey(position.getLedgerId(), - position.getEntryId())) { - ackOwnedConsumer = consumer; - break; - } - } - } else { - ackOwnedConsumer = this; + private boolean removePendingAcks(Consumer ackOwnedConsumer, Position position) { + if (!ackOwnedConsumer.getPendingAcks().remove(position.getLedgerId(), position.getEntryId())) { + // Message was already removed by the other consumer + return false; } - - // remove pending message from appropriate consumer and unblock unAckMsg-flow if requires - LongPair ackedPosition = ackOwnedConsumer != null - ? ackOwnedConsumer.getPendingAcks().get(position.getLedgerId(), position.getEntryId()) - : null; - if (ackedPosition != null) { - if (!ackOwnedConsumer.getPendingAcks().remove(position.getLedgerId(), position.getEntryId())) { - // Message was already removed by the other consumer - return false; - } - if (log.isDebugEnabled()) { - log.debug("[{}-{}] consumer {} received ack {}", topicName, subscription, consumerId, position); - } - // unblock consumer-throttling when limit check is disabled or receives half of maxUnackedMessages => - // consumer can start again consuming messages - int unAckedMsgs = UNACKED_MESSAGES_UPDATER.get(ackOwnedConsumer); - if ((((unAckedMsgs <= getMaxUnackedMessages() / 2) && ackOwnedConsumer.blockedConsumerOnUnackedMsgs) - && ackOwnedConsumer.shouldBlockConsumerOnUnackMsgs()) - || !shouldBlockConsumerOnUnackMsgs()) { - ackOwnedConsumer.blockedConsumerOnUnackedMsgs = false; - flowConsumerBlockedPermits(ackOwnedConsumer); - } - return true; + if (log.isDebugEnabled()) { + log.debug("[{}-{}] consumer {} received ack {}", topicName, subscription, consumerId, position); } - return false; + // unblock consumer-throttling when limit check is disabled or receives half of maxUnackedMessages => + // consumer can start again consuming messages + int unAckedMsgs = UNACKED_MESSAGES_UPDATER.get(ackOwnedConsumer); + if ((((unAckedMsgs <= getMaxUnackedMessages() / 2) && ackOwnedConsumer.blockedConsumerOnUnackedMsgs) + && ackOwnedConsumer.shouldBlockConsumerOnUnackMsgs()) + || !shouldBlockConsumerOnUnackMsgs()) { + ackOwnedConsumer.blockedConsumerOnUnackedMsgs = false; + flowConsumerBlockedPermits(ackOwnedConsumer); + } + return true; } public ConcurrentLongLongPairHashMap getPendingAcks() { From 679ac083f5cc7e84897375f2288998b90e63c1cc Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Mon, 29 Jul 2024 17:49:34 +0800 Subject: [PATCH 2/2] rename method --- .../org/apache/pulsar/broker/service/Consumer.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 8af98bbfd45cd..7f46e8969eb53 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -537,10 +537,11 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map ackOwnerConsumerPair = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); - Consumer ackOwnerConsumer = ackOwnerConsumerPair.getLeft(); + Pair ackOwnerConsumerAndBatchSize = + getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(), msgId.getEntryId()); + Consumer ackOwnerConsumer = ackOwnerConsumerAndBatchSize.getLeft(); long ackedCount; - long batchSize = ackOwnerConsumerPair.getRight(); + long batchSize = ackOwnerConsumerAndBatchSize.getRight(); if (msgId.getAckSetsCount() > 0) { long[] ackSets = new long[msgId.getAckSetsCount()]; for (int j = 0; j < msgId.getAckSetsCount(); j++) { @@ -606,8 +607,8 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { for (int i = 0; i < ack.getMessageIdsCount(); i++) { MessageIdData msgId = ack.getMessageIdAt(i); Position position = AckSetStateUtil.createPositionWithAckSet(msgId.getLedgerId(), msgId.getEntryId(), null); - Pair ackOwnerConsumerPair = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId()); - Consumer ackOwnerConsumer = ackOwnerConsumerPair.getLeft(); + Consumer ackOwnerConsumer = getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(), + msgId.getEntryId()).getLeft(); // acked count at least one long ackedCount; long batchSize; @@ -733,7 +734,7 @@ private boolean checkCanRemovePendingAcksAndHandle(Consumer ackOwnedConsumer, * @param entryId The ID of the entry. * @return Pair */ - private Pair getAckOwnerConsumer(long ledgerId, long entryId) { + private Pair getAckOwnerConsumerAndBatchSize(long ledgerId, long entryId) { if (Subscription.isIndividualAckMode(subType)) { LongPair longPair = getPendingAcks().get(ledgerId, entryId); if (longPair != null) {