[improve][broker] Cut PendingAcksMap per-entry overhead with primitive storage

microbench/src/main/java/org/apache/pulsar/broker/service/package-info.java

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Benchmarks for broker service internals.
+ */
+package org.apache.pulsar.broker.service;

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

@@ -83,6 +83,7 @@
 public class Consumer {
 
     private static final Logger LOG = Logger.get(Consumer.class);
+    private static final int PENDING_ACK_NOT_FOUND = -1;
     private final Logger log;
 
     private final Subscription subscription;
@@ -649,10 +650,10 @@ private CompletableFuture<Long> individualAck(CommandAck ack, Map<String, Long>
                         addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
                     }
                 } else if (!hasAckSet) {
-                    IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet(
+                    int removed = ackOwnerConsumer.removePendingAckAndGetRemainingUnacked(
                             position.getLedgerId(), position.getEntryId());
-                    if (removed != null) {
-                        addAndGetUnAckedMsgs(ackOwnerConsumer, -removed.leftInt());
+                    if (removed != PENDING_ACK_NOT_FOUND) {
+                        addAndGetUnAckedMsgs(ackOwnerConsumer, -removed);
                         updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer);
                     }
                 }
@@ -744,10 +745,10 @@ private void applyPendingAckCompletions(List<PendingAckCompletion> pendingAckCom
                     }
                 }
             } else {
-                IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet(
+                int removed = ackOwnerConsumer.removePendingAckAndGetRemainingUnacked(
                         position.getLedgerId(), position.getEntryId());
-                if (removed != null) {
-                    addAndGetUnAckedMsgs(ackOwnerConsumer, -removed.leftInt());
+                if (removed != PENDING_ACK_NOT_FOUND) {
+                    addAndGetUnAckedMsgs(ackOwnerConsumer, -removed);
                 }
             }
             updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer);
@@ -841,16 +842,16 @@ private void checkAckValidationError(CommandAck ack, Position position) {
      */
     private ObjectIntPair<Consumer> getAckOwnerConsumerAndBatchSize(long ledgerId, long entryId) {
         if (Subscription.isIndividualAckMode(subType)) {
-            IntIntPair pendingAck = getPendingAcks().get(ledgerId, entryId);
-            if (pendingAck != null) {
-                return ObjectIntPair.of(this, pendingAck.leftInt());
+            int remainingUnacked = getPendingAcks().getRemainingUnacked(ledgerId, entryId);
+            if (remainingUnacked != PENDING_ACK_NOT_FOUND) {
+                return ObjectIntPair.of(this, remainingUnacked);
             } 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) {
-                        pendingAck = consumer.getPendingAcks().get(ledgerId, entryId);
-                        if (pendingAck != null) {
-                            return ObjectIntPair.of(consumer, pendingAck.leftInt());
+                        remainingUnacked = consumer.getPendingAcks().getRemainingUnacked(ledgerId, entryId);
+                        if (remainingUnacked != PENDING_ACK_NOT_FOUND) {
+                            return ObjectIntPair.of(consumer, remainingUnacked);
                         }
                     }
                 }
@@ -1207,6 +1208,20 @@ public IntIntPair removePendingAckAndGet(long ledgerId, long entryId) {
         return null;
     }
 
+    /**
+     * Atomically remove the pending ack entry and return its remaining unacked count.
+     *
+     * <p>No-op if {@code pendingAcks} is not initialized.
+     *
+     * @return the remaining unacked count, or {@code -1} if not found
+     */
+    public int removePendingAckAndGetRemainingUnacked(long ledgerId, long entryId) {
+        if (pendingAcks != null) {
+            return pendingAcks.removeAndGetRemainingUnacked(ledgerId, entryId);
+        }
+        return PENDING_ACK_NOT_FOUND;
+    }
+
     /**
      * Remove all pending acks up to the given mark-delete position and decrement the consumer's unacked message
      * counter by the remaining unacked count for each removed entry.
@@ -1273,9 +1288,10 @@ public void redeliverUnacknowledgedMessages(List<MessageIdData> messageIds) {
         List<Position> pendingPositions = new ArrayList<>();
         for (MessageIdData msg : messageIds) {
             Position position = PositionFactory.create(msg.getLedgerId(), msg.getEntryId());
-            IntIntPair pendingAck = pendingAcks.removeAndGet(position.getLedgerId(), position.getEntryId());
-            if (pendingAck != null) {
-                totalRedeliveryMessages += pendingAck.leftInt();
+            int remainingUnacked = pendingAcks.removeAndGetRemainingUnacked(
+                    position.getLedgerId(), position.getEntryId());
+            if (remainingUnacked != PENDING_ACK_NOT_FOUND) {
+                totalRedeliveryMessages += remainingUnacked;
                 pendingPositions.add(position);
             }
         }