[improve][txn] change delete pending ack position from foreach to fir…

…stKey (apache#16927)
nodece · Aug 4, 2022 · 5b65fda · 5b65fda
1 parent b395ad4
commit 5b65fda
Show file tree

Hide file tree

Showing 2 changed files with 60 additions and 6 deletions.
diff --git a/.../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java b/.../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
@@ -95,7 +95,7 @@ public class PendingAckHandleImpl extends PendingAckHandleState implements Pendi
      *     <p>
      *         If it does not exits the map, the position will be added to the map.
      */
-    private Map<PositionImpl, MutablePair<PositionImpl, Integer>> individualAckPositions;
+    private ConcurrentSkipListMap<PositionImpl, MutablePair<PositionImpl, Integer>> individualAckPositions;
 
     /**
      * The map is for transaction with position witch was cumulative acked by this transaction.
@@ -884,12 +884,14 @@ public synchronized void clearIndividualPosition(Position position) {
             individualAckPositions.remove(position);
         }
 
-        individualAckPositions.forEach((persistentPosition, positionIntegerMutablePair) -> {
-            if (persistentPosition.compareTo((PositionImpl) persistentSubscription
+        while (individualAckPositions.firstEntry() != null) {
+            if (individualAckPositions.firstKey().compareTo((PositionImpl) persistentSubscription
                     .getCursor().getMarkDeletedPosition()) < 0) {
-                individualAckPositions.remove(persistentPosition);
+                individualAckPositions.remove(individualAckPositions.firstKey());
+            } else {
+                break;
             }
-        });
+        }
     }
 
     @Override

diff --git a/...st/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java b/...st/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java
@@ -21,10 +21,10 @@
 
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
-
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.commons.collections4.map.LinkedMap;
+import org.apache.commons.lang3.tuple.MutablePair;
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
@@ -40,13 +40,15 @@
 import org.apache.pulsar.common.util.collections.BitSetRecyclable;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.awaitility.Awaitility;
+import org.powermock.reflect.Whitebox;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 import java.lang.reflect.Field;
 import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentSkipListMap;
@@ -280,6 +282,56 @@ public void txnAckTestBatchAndSharedSubMemoryDeleteTest() throws Exception {
         }
     }
 
+    @Test
+    public void testPendingAckClearPositionIsSmallerThanMarkDelete() throws Exception {
+        String normalTopic = NAMESPACE1 + "/testPendingAckClearPositionIsSmallerThanMarkDelete";
+        String subscriptionName = "test";
+
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer()
+                .topic(normalTopic)
+                .subscriptionName(subscriptionName)
+                .enableBatchIndexAcknowledgment(true)
+                .subscriptionType(SubscriptionType.Shared)
+                .subscribe();
+
+        @Cleanup
+        Producer<byte[]> producer = pulsarClient.newProducer()
+                .topic(normalTopic)
+                .enableBatching(true)
+                .batchingMaxMessages(200)
+                .create();
+
+        // mark delete position
+        producer.send("test1".getBytes());
+
+        Transaction commitTxn = getTxn();
+
+        consumer.acknowledgeAsync(consumer.receive().getMessageId(), commitTxn).get();
+
+        PendingAckHandle pendingAckHandle = Whitebox.getInternalState(getPulsarServiceList().get(0)
+                .getBrokerService().getTopic("persistent://" + normalTopic, false).get().get()
+                .getSubscription(subscriptionName), "pendingAckHandle");
+
+        Map<PositionImpl, MutablePair<PositionImpl, Integer>> individualAckPositions =
+                Whitebox.getInternalState(pendingAckHandle, "individualAckPositions");
+        // one message in pending ack state
+        assertEquals(1, individualAckPositions.size());
+
+        // put the PositionImpl.EARLIEST to the map
+        individualAckPositions.put(PositionImpl.EARLIEST, new MutablePair<>(PositionImpl.EARLIEST, 0));
+
+        // put the PositionImpl.LATEST to the map
+        individualAckPositions.put(PositionImpl.LATEST, new MutablePair<>(PositionImpl.EARLIEST, 0));
+
+        // three position in pending ack state
+        assertEquals(3, individualAckPositions.size());
+
+        // commit this txn will delete the received position and PositionImpl.EARLIEST, don't delete PositionImpl.LATEST
+        commitTxn.commit().get();
+        assertEquals(1, individualAckPositions.size());
+    }
+
     private Transaction getTxn() throws Exception {
         return pulsarClient
                 .newTransaction()