You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/17 06:12:30 UTC
(pulsar) 11/14: [improve][broker] Keep each ledger's pending acks in sorted blocks instead of an AVL tree (#26955)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d79b840ad08a43823d3e9cdfdff1e43b6431b860
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Oct 16 21:42:20 2026 -0700
[improve][broker] Keep each ledger's pending acks in sorted blocks instead of an AVL tree (#26955)
(cherry picked from commit 8fd72345a2a389e07c29cd10b23c46c73d4cbbaa)
---
.../service/PendingAcksLedgerMapBenchmark.java | 373 ++++++++++++++
.../service/PendingAcksMapOneEntryBenchmark.java | 62 +++
.../broker/service/PendingAcksLedgerMap.java | 535 +++++++++++++++++++++
.../pulsar/broker/service/PendingAcksMap.java | 103 ++--
.../broker/service/PendingAcksLedgerMapTest.java | 501 +++++++++++++++++++
.../pulsar/broker/service/PendingAcksMapTest.java | 179 +++++++
6 files changed, 1708 insertions(+), 45 deletions(-)
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapBenchmark.java
new file mode 100644
index 000000000000..4d193dde5ceb
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapBenchmark.java
@@ -0,0 +1,373 @@
+/*
+ * 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.
+ */
+package org.apache.pulsar.broker.service;
+
+import it.unimi.dsi.fastutil.longs.Long2LongAVLTreeMap;
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.objects.ObjectBidirectionalIterator;
+import java.util.Arrays;
+import java.util.SplittableRandom;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Compares the per-ledger structures of a consumer's pending acks: the AVL tree that {@link PendingAcksMap} used
+ * before, {@code Long2LongAVLTreeMap}, and {@link PendingAcksLedgerMap}, in the patterns of dispatch and acks. It
+ * excludes {@link PendingAcksMap}'s lock and outer ledger map, which are the same for both and which
+ * {@link PendingAckMapCycleBenchmark} includes.
+ *
+ * <p>The window is the number of pending acks. Every operation that removes an entry adds a new one after the others,
+ * as a dispatch does, so that the window stays the same. The results are per entry, except for {@code traverse},
+ * which is per traversal of the window.
+ *
+ * <p>{@link PendingAcksMap} reuses an emptied ledger map, where it created a new AVL tree before:
+ * {@code oneEntryAtATime} does the same with each structure.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class PendingAcksLedgerMapBenchmark {
+ // the entries that an ack command acknowledges together, and that a mark-delete removes together
+ private static final int GROUP = 64;
+ // one in this many acked entries is redelivered: removed and added again later, out of order
+ private static final int REDELIVERY_INTERVAL = 16;
+ // a redelivered entry is added again this many acks later
+ private static final int REDELIVERY_DELAY = 64;
+ // the number of maps in tinyMaps, each with a window of TINY_WINDOW entries
+ private static final int TINY_MAPS = 8192;
+ private static final int TINY_WINDOW = 8;
+ // the length of the precomputed sequence of random indexes that ackShuffled acks
+ private static final int RANDOM_INDEXES = 1 << 20;
+
+ /** The operations of a ledger's pending acks that the patterns use. */
+ interface LedgerMap {
+ long get(long entryId);
+
+ void put(long entryId, long value);
+
+ long remove(long entryId);
+
+ /** Removes the entries up to the entry ID, inclusive, as a mark-delete does; returns their count. */
+ int removeUpTo(long entryId);
+
+ void forEach(Blackhole blackhole);
+ }
+
+ /** The previous structure, as {@link PendingAcksMap} used it. */
+ static final class AvlTreeLedgerMap implements LedgerMap {
+ private final Long2LongAVLTreeMap map = new Long2LongAVLTreeMap();
+
+ AvlTreeLedgerMap() {
+ map.defaultReturnValue(PendingAckValues.PACKED_NOT_FOUND);
+ }
+
+ @Override
+ public long get(long entryId) {
+ return map.get(entryId);
+ }
+
+ @Override
+ public void put(long entryId, long value) {
+ map.put(entryId, value);
+ }
+
+ @Override
+ public long remove(long entryId) {
+ return map.remove(entryId);
+ }
+
+ @Override
+ public int removeUpTo(long entryId) {
+ int removed = 0;
+ ObjectBidirectionalIterator<Long2LongMap.Entry> iterator =
+ map.headMap(entryId + 1).long2LongEntrySet().iterator();
+ while (iterator.hasNext()) {
+ iterator.next();
+ iterator.remove();
+ removed++;
+ }
+ return removed;
+ }
+
+ @Override
+ public void forEach(Blackhole blackhole) {
+ for (Long2LongMap.Entry entry : map.long2LongEntrySet()) {
+ blackhole.consume(entry.getLongKey());
+ blackhole.consume(entry.getLongValue());
+ }
+ }
+ }
+
+ /** The new structure, as {@link PendingAcksMap} uses it. */
+ static final class BlockListLedgerMap implements LedgerMap {
+ private final PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+
+ @Override
+ public long get(long entryId) {
+ return map.get(entryId);
+ }
+
+ @Override
+ public void put(long entryId, long value) {
+ map.put(entryId, value);
+ }
+
+ @Override
+ public long remove(long entryId) {
+ return map.remove(entryId);
+ }
+
+ @Override
+ public int removeUpTo(long entryId) {
+ // as PendingAcksMap.removeAllUpTo does, which tolerates callbacks that change the map
+ int removed = 0;
+ long next = map.hasEntriesUpTo(entryId) ? map.firstEntryId() : PendingAcksLedgerMap.NO_ENTRY;
+ while (next != PendingAcksLedgerMap.NO_ENTRY && next <= entryId) {
+ if (!PendingAckValues.isNotFound(map.get(next))) {
+ map.remove(next);
+ removed++;
+ }
+ next = map.higherEntryId(next);
+ }
+ return removed;
+ }
+
+ @Override
+ public void forEach(Blackhole blackhole) {
+ map.forEach((entryId, value) -> {
+ blackhole.consume(entryId);
+ blackhole.consume(value);
+ });
+ }
+ }
+
+ @Param({"AVL_TREE", "BLOCK_LIST"})
+ public String implementation;
+
+ @Param({"1024", "65536"})
+ public int window;
+
+ private final long packedValue = PendingAckValues.pack(1, 0);
+
+ private LedgerMap map;
+ // the window's entry IDs; an operation acks the entry at an index and replaces it with a new one
+ private long[] entryIds;
+ private long nextEntryId;
+ private int cursor;
+ // the negatively acked entry IDs, added again REDELIVERY_DELAY acks after their removal
+ private long[] negativelyAcked;
+ // the redelivered entry IDs, acked REDELIVERY_DELAY acks after they were added again
+ private long[] redelivered;
+ private int acks;
+
+ private LedgerMap[] tinyMaps;
+ private long[] tinyNextEntryIds;
+ // the index of a random entry of the window, for each ack of ackShuffled
+ private int[] randomIndexes;
+ // the emptied map that oneEntryAtATime reuses with the block list
+ private LedgerMap emptyMap;
+ private long nextSingleEntryId;
+
+ private LedgerMap newLedgerMap() {
+ return "AVL_TREE".equals(implementation) ? new AvlTreeLedgerMap() : new BlockListLedgerMap();
+ }
+
+ @Setup(Level.Trial)
+ public void setup() {
+ map = newLedgerMap();
+ entryIds = new long[window];
+ for (int i = 0; i < window; i++) {
+ entryIds[i] = i;
+ map.put(i, packedValue);
+ }
+ nextEntryId = window;
+ negativelyAcked = new long[REDELIVERY_DELAY];
+ Arrays.fill(negativelyAcked, -1);
+ redelivered = new long[REDELIVERY_DELAY];
+ Arrays.fill(redelivered, -1);
+ tinyMaps = new LedgerMap[TINY_MAPS];
+ tinyNextEntryIds = new long[TINY_MAPS];
+ for (int m = 0; m < TINY_MAPS; m++) {
+ tinyMaps[m] = newLedgerMap();
+ for (int i = 0; i < TINY_WINDOW; i++) {
+ tinyMaps[m].put(i, packedValue);
+ }
+ tinyNextEntryIds[m] = TINY_WINDOW;
+ }
+ SplittableRandom random = new SplittableRandom(1);
+ randomIndexes = new int[RANDOM_INDEXES];
+ for (int i = 0; i < RANDOM_INDEXES; i++) {
+ randomIndexes[i] = random.nextInt(window);
+ }
+ emptyMap = newLedgerMap();
+ }
+
+ /** The index of the next entry to ack: in order, or a random one of the window. */
+ private int nextIndex(boolean shuffled) {
+ int next = cursor++;
+ return shuffled ? randomIndexes[next & (RANDOM_INDEXES - 1)] : next & (window - 1);
+ }
+
+ // Acks the entry at the index: looks it up and removes it, and dispatches a new entry in its place
+ private long ackAndReplace(int index) {
+ long entryId = entryIds[index];
+ long value = map.get(entryId);
+ map.remove(entryId);
+ long added = nextEntryId++;
+ map.put(added, packedValue);
+ entryIds[index] = added;
+ return value;
+ }
+
+ /** Acks arrive in the order of the entries, the common case: lookup, removal and a new entry. */
+ @Benchmark
+ public long ackInOrder() {
+ return ackAndReplace(nextIndex(false));
+ }
+
+ /**
+ * Acks arrive in a random order over the window, as when a consumer processes messages at different speeds: each
+ * ack is for a random entry of the window, which is replaced by a new one.
+ */
+ @Benchmark
+ public long ackShuffled() {
+ return ackAndReplace(nextIndex(true));
+ }
+
+ /**
+ * An ack command acknowledges a group of entries in order: they are all looked up before any of them is removed
+ * after the ack is persisted, and new entries are dispatched after that.
+ */
+ @Benchmark
+ @OperationsPerInvocation(GROUP)
+ public long ackGroupInOrder() {
+ int first = cursor;
+ cursor += GROUP;
+ long sum = 0;
+ for (int i = 0; i < GROUP; i++) {
+ sum += map.get(entryIds[(first + i) & (window - 1)]);
+ }
+ for (int i = 0; i < GROUP; i++) {
+ map.remove(entryIds[(first + i) & (window - 1)]);
+ }
+ for (int i = 0; i < GROUP; i++) {
+ long added = nextEntryId++;
+ map.put(added, packedValue);
+ entryIds[(first + i) & (window - 1)] = added;
+ }
+ return sum;
+ }
+
+ /**
+ * Acks in order, where one in {@value #REDELIVERY_INTERVAL} entries is negatively acked: it is removed, added again
+ * {@value #REDELIVERY_DELAY} acks later, older than the others, and acked another {@value #REDELIVERY_DELAY} acks
+ * later. The result is per ack of a dispatched entry, including the redeliveries' share.
+ */
+ @Benchmark
+ public long ackInOrderWithRedeliveries() {
+ int slot = acks++ & (REDELIVERY_DELAY - 1);
+ long value = 0;
+ if (redelivered[slot] >= 0) {
+ // the redelivered entry is acked
+ value += map.get(redelivered[slot]);
+ map.remove(redelivered[slot]);
+ redelivered[slot] = -1;
+ }
+ if (negativelyAcked[slot] >= 0) {
+ // the negatively acked entry is dispatched again, out of order
+ map.put(negativelyAcked[slot], packedValue);
+ redelivered[slot] = negativelyAcked[slot];
+ negativelyAcked[slot] = -1;
+ }
+ int index = nextIndex(false);
+ long entryId = entryIds[index];
+ value += map.get(entryId);
+ map.remove(entryId);
+ if (acks % REDELIVERY_INTERVAL == 0) {
+ // negatively acked instead
+ negativelyAcked[slot] = entryId;
+ }
+ long added = nextEntryId++;
+ map.put(added, packedValue);
+ entryIds[index] = added;
+ return value;
+ }
+
+ /** A mark-delete removes the oldest entries up to a position, and new entries are dispatched. */
+ @Benchmark
+ @OperationsPerInvocation(GROUP)
+ public int removeUpToMarkDelete() {
+ long upTo = nextEntryId - window + GROUP - 1;
+ int removed = map.removeUpTo(upTo);
+ for (int i = 0; i < GROUP; i++) {
+ map.put(nextEntryId++, packedValue);
+ }
+ return removed;
+ }
+
+ /** A traversal of the window, as a redelivery of all pending acks or a consumer's close does. */
+ @Benchmark
+ public void traverse(Blackhole blackhole) {
+ map.forEach(blackhole);
+ }
+
+ /**
+ * A consumer whose pending acks drop to none between entries, as on a topic with a low rate: the ledger's map is
+ * removed when it empties, and a map is taken for the next entry. {@link PendingAcksMap} created a new AVL tree
+ * for it, and reuses the emptied block list.
+ */
+ @Benchmark
+ public long oneEntryAtATime() {
+ LedgerMap single = "AVL_TREE".equals(implementation) ? new AvlTreeLedgerMap() : emptyMap;
+ long entryId = nextSingleEntryId++;
+ single.put(entryId, packedValue);
+ long value = single.get(entryId);
+ single.remove(entryId);
+ return value;
+ }
+
+ /** Many consumers with a few pending acks each, as with many slow or idle consumers: an ack on a random one. */
+ @Benchmark
+ public long tinyMaps() {
+ int m = (cursor++ * 4051) & (TINY_MAPS - 1);
+ LedgerMap tiny = tinyMaps[m];
+ long next = tinyNextEntryIds[m]++;
+ long entryId = next - TINY_WINDOW;
+ long value = tiny.get(entryId);
+ tiny.remove(entryId);
+ tiny.put(next, packedValue);
+ return value;
+ }
+}
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksMapOneEntryBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksMapOneEntryBenchmark.java
new file mode 100644
index 000000000000..6cf77b76c3e3
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/service/PendingAcksMapOneEntryBenchmark.java
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.apache.pulsar.broker.service;
+
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * A consumer whose pending acks drop to none between messages, as on a topic with a low rate, with the real
+ * pending-ACK map's locks and outer ledger map: each message is added, looked up when its ack arrives and removed,
+ * which empties the map.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class PendingAcksMapOneEntryBenchmark {
+ private PendingAcksMap pendingAcks;
+ private long nextEntryId;
+
+ @Setup
+ public void setup() {
+ Consumer owner = new Consumer("benchmark-owner", 0);
+ pendingAcks = new PendingAcksMap(owner, () -> null, () -> null);
+ }
+
+ @Benchmark
+ public int oneEntryAtATime() {
+ long entryId = nextEntryId++;
+ pendingAcks.addPendingAckIfAllowed(1, entryId, 1, 0);
+ int remainingUnacked = pendingAcks.getRemainingUnacked(1, entryId);
+ pendingAcks.removeAndGetRemainingUnacked(1, entryId);
+ return remainingUnacked;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMap.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMap.java
new file mode 100644
index 000000000000..2a0948bd86e0
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksLedgerMap.java
@@ -0,0 +1,535 @@
+/*
+ * 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.
+ */
+package org.apache.pulsar.broker.service;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+/**
+ * The pending acks of one ledger: a map from entry ID to a packed {@link PendingAckValues} value, in the order of the
+ * entry IDs. Not thread-safe; {@link PendingAcksMap} guards it with its lock.
+ *
+ * <p>The entries are kept in a list of blocks, each a sorted array of up to {@value #BLOCK_CAPACITY} entries, in the
+ * order of their entry IDs, with each block's entries in a window of its arrays. A consumer is sent a ledger's entries
+ * mostly in increasing order and acks them mostly in the same order, so that adding an entry appends it to the last
+ * block and removing one advances the first block's window; a lookup is a binary search over the blocks' first
+ * entry IDs, kept in one array, and over one block's consecutive memory, without the pointer chasing of a tree. An
+ * entry added out of order, as a redelivery adds it, shifts the entries of one block at most, which splits when it is
+ * full. Empty blocks are dropped and a block that becomes sparse merges with its neighbor, so that the memory follows
+ * the number of entries. A lone block starts small, grows as entries are added and shrinks as they are removed, so
+ * that a map of a few entries stays small. While the map has more than one block, the last emptied block is kept for
+ * the next one, so that a window of entries that moves through the map doesn't allocate.
+ *
+ * <p>The methods that read the map don't change it: {@link PendingAcksMap} calls them concurrently under its read
+ * lock.
+ */
+final class PendingAcksLedgerMap {
+ static final int BLOCK_CAPACITY = 128;
+ // The smallest capacity of a block: a lone block starts with it, doubles up to BLOCK_CAPACITY as entries are added
+ // and halves as they are removed
+ static final int INITIAL_BLOCK_CAPACITY = 16;
+ // A block with fewer entries than this merges with a neighbor when they fit in half a block together
+ private static final int MERGE_THRESHOLD = BLOCK_CAPACITY / 4;
+
+ private static final class Block {
+ long[] entryIds;
+ long[] values;
+ // the entries are in [start, end)
+ int start;
+ int end;
+
+ Block(int capacity) {
+ entryIds = new long[capacity];
+ values = new long[capacity];
+ }
+
+ int capacity() {
+ return entryIds.length;
+ }
+
+ int size() {
+ return end - start;
+ }
+
+ long firstEntryId() {
+ return entryIds[start];
+ }
+
+ long lastEntryId() {
+ return entryIds[end - 1];
+ }
+
+ int indexOf(long entryId) {
+ return Arrays.binarySearch(entryIds, start, end, entryId);
+ }
+
+ /** Inserts at the insertion point, which keeps the order; the block's arrays mustn't be full. */
+ void insertAt(int insertionPoint, long entryId, long value) {
+ boolean roomBefore = start > 0;
+ boolean roomAfter = end < capacity();
+ if (roomBefore && (!roomAfter || insertionPoint - start < end - insertionPoint)) {
+ // shift the entries before the insertion point to the left
+ System.arraycopy(entryIds, start, entryIds, start - 1, insertionPoint - start);
+ System.arraycopy(values, start, values, start - 1, insertionPoint - start);
+ start--;
+ entryIds[insertionPoint - 1] = entryId;
+ values[insertionPoint - 1] = value;
+ } else {
+ System.arraycopy(entryIds, insertionPoint, entryIds, insertionPoint + 1, end - insertionPoint);
+ System.arraycopy(values, insertionPoint, values, insertionPoint + 1, end - insertionPoint);
+ end++;
+ entryIds[insertionPoint] = entryId;
+ values[insertionPoint] = value;
+ }
+ }
+
+ void removeAt(int index) {
+ if (index - start < end - 1 - index) {
+ System.arraycopy(entryIds, start, entryIds, start + 1, index - start);
+ System.arraycopy(values, start, values, start + 1, index - start);
+ start++;
+ } else {
+ System.arraycopy(entryIds, index + 1, entryIds, index, end - 1 - index);
+ System.arraycopy(values, index + 1, values, index, end - 1 - index);
+ end--;
+ }
+ if (start == end) {
+ start = 0;
+ end = 0;
+ }
+ }
+
+ /** Moves the entries to the start of the arrays. */
+ void compact() {
+ if (start > 0) {
+ System.arraycopy(entryIds, start, entryIds, 0, size());
+ System.arraycopy(values, start, values, 0, size());
+ end -= start;
+ start = 0;
+ }
+ }
+
+ /** Grows the arrays to at least the capacity, doubling, up to {@link #BLOCK_CAPACITY}; compacts them. */
+ void grow(int minCapacity) {
+ int capacity = capacity();
+ while (capacity < minCapacity) {
+ capacity *= 2;
+ }
+ resize(Math.min(capacity, BLOCK_CAPACITY));
+ }
+
+ /** Moves the entries to new arrays of the capacity, which must hold them. */
+ void resize(int capacity) {
+ int size = size();
+ entryIds = Arrays.copyOfRange(entryIds, start, start + capacity);
+ values = Arrays.copyOfRange(values, start, start + capacity);
+ start = 0;
+ end = size;
+ }
+ }
+
+ /** A callback for each entry, in the order of the entry IDs. */
+ interface EntryConsumer {
+ void accept(long entryId, long value);
+ }
+
+ /**
+ * Returned by {@link #higherEntryId} when there's no entry after the entry ID. It can't be an entry ID, since
+ * BookKeeper's entry IDs aren't negative.
+ */
+ static final long NO_ENTRY = Long.MIN_VALUE;
+
+ private final ArrayList<Block> blocks = new ArrayList<>(1);
+ // the first entry ID of each block, in step with the blocks, for the binary search over the blocks
+ private long[] firstEntryIds = new long[1];
+ // The last emptied block, reused for the next block: one of the full capacity while the map has more than one
+ // block, so that a window that moves through the map doesn't allocate, or the lone block of the smallest capacity
+ // that an empty map had, so that a map that empties between entries, which PendingAcksMap reuses, doesn't allocate
+ private Block spareBlock;
+ // changed when an entry is added or removed, so that a traversal notices when its callback changed the map
+ private int modCount;
+ // the most blocks since the list's capacity was last trimmed, to release the capacity after the map shrinks
+ private int peakBlocks;
+ private int size;
+ // the blocks allocated by this map, for tests
+ private int allocatedBlocks;
+
+ int size() {
+ return size;
+ }
+
+ boolean isEmpty() {
+ return size == 0;
+ }
+
+ @VisibleForTesting
+ int blockCount() {
+ return blocks.size();
+ }
+
+ @VisibleForTesting
+ int allocatedBlocks() {
+ return allocatedBlocks;
+ }
+
+ @VisibleForTesting
+ int blockCapacity(int blockIndex) {
+ return blocks.get(blockIndex).capacity();
+ }
+
+ /** Checks the map's internal consistency, for tests. */
+ @VisibleForTesting
+ void checkInvariants() {
+ int entries = 0;
+ long previous = Long.MIN_VALUE;
+ for (int b = 0; b < blocks.size(); b++) {
+ Block block = blocks.get(b);
+ if (block.size() == 0) {
+ throw new IllegalStateException("empty block " + b);
+ }
+ if (firstEntryIds[b] != block.firstEntryId()) {
+ throw new IllegalStateException("first entry ID of block " + b + " is " + block.firstEntryId()
+ + ", indexed as " + firstEntryIds[b]);
+ }
+ for (int i = block.start; i < block.end; i++) {
+ if (block.entryIds[i] <= previous) {
+ throw new IllegalStateException("entry " + block.entryIds[i] + " after " + previous);
+ }
+ previous = block.entryIds[i];
+ }
+ entries += block.size();
+ }
+ if (entries != size) {
+ throw new IllegalStateException(entries + " entries, size " + size);
+ }
+ if (spareBlock != null && spareBlock.size() != 0) {
+ throw new IllegalStateException("a spare block with entries");
+ }
+ if (spareBlock != null && (blocks.size() == 1
+ || blocks.isEmpty() && spareBlock.capacity() != INITIAL_BLOCK_CAPACITY)) {
+ throw new IllegalStateException("a spare block of capacity " + spareBlock.capacity() + " with "
+ + blocks.size() + " blocks");
+ }
+ if (blocks.size() > 1) {
+ for (Block block : blocks) {
+ if (block.capacity() != BLOCK_CAPACITY) {
+ throw new IllegalStateException("a block of capacity " + block.capacity() + " among others");
+ }
+ }
+ }
+ }
+
+ /** The entry's value, or {@link PendingAckValues#PACKED_NOT_FOUND}. */
+ long get(long entryId) {
+ if (isFirst(entryId)) {
+ return firstValue();
+ }
+ int blockIndex = blockIndexFor(entryId);
+ if (blockIndex < 0) {
+ return PendingAckValues.PACKED_NOT_FOUND;
+ }
+ Block block = blocks.get(blockIndex);
+ int index = block.indexOf(entryId);
+ return index >= 0 ? block.values[index] : PendingAckValues.PACKED_NOT_FOUND;
+ }
+
+ // Whether the entry is the first one: acks mostly come in the order of the entries, and both the ack's lookup and
+ // its removal then find it without a search
+ private boolean isFirst(long entryId) {
+ return size > 0 && firstEntryId() == entryId;
+ }
+
+ boolean containsKey(long entryId) {
+ return !PendingAckValues.isNotFound(get(entryId));
+ }
+
+ /**
+ * Sets the entry's value, which mustn't be {@link PendingAckValues#PACKED_NOT_FOUND}.
+ *
+ * @return the previous value, or {@link PendingAckValues#PACKED_NOT_FOUND}
+ */
+ long put(long entryId, long value) {
+ Block last = blocks.isEmpty() ? null : blocks.get(blocks.size() - 1);
+ if (last == null || entryId > last.lastEntryId()) {
+ // the common case: an entry after the others
+ boolean newBlock = false;
+ if (last == null || last.end == last.capacity()) {
+ if (last != null && last.start >= last.capacity() / 4) {
+ // compacting frees a quarter of the block at least, so that it doesn't happen on every append
+ last.compact();
+ } else if (last != null && last.capacity() < BLOCK_CAPACITY) {
+ last.grow(last.capacity() + 1);
+ } else {
+ // a map's first block starts small, the blocks after it as full ones
+ last = newBlock(last == null ? INITIAL_BLOCK_CAPACITY : BLOCK_CAPACITY);
+ newBlock = true;
+ }
+ }
+ last.entryIds[last.end] = entryId;
+ last.values[last.end] = value;
+ last.end++;
+ if (newBlock) {
+ addBlock(blocks.size(), last);
+ }
+ size++;
+ modCount++;
+ return PendingAckValues.PACKED_NOT_FOUND;
+ }
+ // the first block whose entries reach the entry ID
+ int blockIndex = Math.max(blockIndexFor(entryId), 0);
+ Block block = blocks.get(blockIndex);
+ int index = block.indexOf(entryId);
+ if (index >= 0) {
+ long previous = block.values[index];
+ block.values[index] = value;
+ return previous;
+ }
+ if (block.size() == BLOCK_CAPACITY) {
+ Block upper = split(blockIndex);
+ if (entryId > block.lastEntryId()) {
+ block = upper;
+ blockIndex++;
+ }
+ index = block.indexOf(entryId);
+ } else if (block.size() == block.capacity()) {
+ // a small first block that is full
+ block.grow(block.capacity() + 1);
+ index = block.indexOf(entryId);
+ }
+ block.insertAt(-index - 1, entryId, value);
+ firstEntryIds[blockIndex] = block.firstEntryId();
+ size++;
+ modCount++;
+ return PendingAckValues.PACKED_NOT_FOUND;
+ }
+
+ /**
+ * Removes the entry.
+ *
+ * @return its value, or {@link PendingAckValues#PACKED_NOT_FOUND} if it wasn't in the map
+ */
+ long remove(long entryId) {
+ if (isFirst(entryId)) {
+ long previous = firstValue();
+ removeFirst();
+ return previous;
+ }
+ int blockIndex = blockIndexFor(entryId);
+ if (blockIndex < 0) {
+ return PendingAckValues.PACKED_NOT_FOUND;
+ }
+ Block block = blocks.get(blockIndex);
+ int index = block.indexOf(entryId);
+ if (index < 0) {
+ return PendingAckValues.PACKED_NOT_FOUND;
+ }
+ long previous = block.values[index];
+ removeAt(blockIndex, block, index);
+ return previous;
+ }
+
+ /** The first entry's ID; the map mustn't be empty. */
+ long firstEntryId() {
+ return firstEntryIds[0];
+ }
+
+ /** The first entry's value; the map mustn't be empty. */
+ long firstValue() {
+ Block first = blocks.get(0);
+ return first.values[first.start];
+ }
+
+ /** Removes the first entry; the map mustn't be empty. */
+ void removeFirst() {
+ Block first = blocks.get(0);
+ removeAt(0, first, first.start);
+ }
+
+ /** Whether the map has an entry with an ID up to {@code maxEntryId}, inclusive. */
+ boolean hasEntriesUpTo(long maxEntryId) {
+ return size > 0 && firstEntryId() <= maxEntryId;
+ }
+
+ /**
+ * Calls the consumer for each entry, in the order of the entry IDs. The consumer may add and remove entries: the
+ * traversal continues with the first entry after the current one as the map is then, so it visits the entries
+ * added after the current one and not those added before it, and never an entry that was removed.
+ */
+ void forEach(EntryConsumer consumer) {
+ for (int b = 0; b < blocks.size(); b++) {
+ Block block = blocks.get(b);
+ for (int i = block.start; i < block.end; i++) {
+ long entryId = block.entryIds[i];
+ int expectedModCount = modCount;
+ consumer.accept(entryId, block.values[i]);
+ if (modCount != expectedModCount) {
+ // the entries may have moved: continue after the entry, wherever the next one is now
+ for (long next = higherEntryId(entryId); next != NO_ENTRY; next = higherEntryId(next)) {
+ consumer.accept(next, get(next));
+ }
+ return;
+ }
+ }
+ }
+ }
+
+ /** The ID of the first entry after the entry ID, or {@link #NO_ENTRY}. */
+ long higherEntryId(long entryId) {
+ if (size == 0) {
+ return NO_ENTRY;
+ }
+ if (firstEntryId() > entryId) {
+ // the common case when entries are removed in order
+ return firstEntryId();
+ }
+ int blockIndex = blockIndexFor(entryId);
+ Block block = blocks.get(blockIndex);
+ int index = block.indexOf(entryId);
+ int next = index >= 0 ? index + 1 : -index - 1;
+ if (next < block.end) {
+ return block.entryIds[next];
+ }
+ return blockIndex + 1 < blocks.size() ? firstEntryIds[blockIndex + 1] : NO_ENTRY;
+ }
+
+ /** The index of the last block whose first entry ID is at most the entry ID, or -1. */
+ private int blockIndexFor(long entryId) {
+ int low = 0;
+ int high = blocks.size() - 1;
+ int found = -1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ if (firstEntryIds[mid] <= entryId) {
+ found = mid;
+ low = mid + 1;
+ } else {
+ high = mid - 1;
+ }
+ }
+ return found;
+ }
+
+ /** A block for the next entries: the spare one, or a new one of the capacity. */
+ private Block newBlock(int capacity) {
+ Block block = spareBlock;
+ if (block != null && block.capacity() >= capacity) {
+ spareBlock = null;
+ return block;
+ }
+ allocatedBlocks++;
+ return new Block(capacity);
+ }
+
+ /** Adds a block that has entries at the index, with its first entry ID. */
+ private void addBlock(int blockIndex, Block block) {
+ int count = blocks.size();
+ if (count == firstEntryIds.length) {
+ firstEntryIds = Arrays.copyOf(firstEntryIds, count * 2);
+ }
+ System.arraycopy(firstEntryIds, blockIndex, firstEntryIds, blockIndex + 1, count - blockIndex);
+ firstEntryIds[blockIndex] = block.firstEntryId();
+ blocks.add(blockIndex, block);
+ peakBlocks = Math.max(peakBlocks, blocks.size());
+ }
+
+ /** Removes the block at the index, keeping it as the spare one. */
+ private void removeBlock(int blockIndex) {
+ Block block = blocks.remove(blockIndex);
+ System.arraycopy(firstEntryIds, blockIndex + 1, firstEntryIds, blockIndex, blocks.size() - blockIndex);
+ // a block merged into another one still has its window
+ block.start = 0;
+ block.end = 0;
+ spareBlock = block;
+ }
+
+ /** Splits a full block in two halves; returns the new upper half, which follows it in the list. */
+ private Block split(int blockIndex) {
+ Block block = blocks.get(blockIndex);
+ Block upper = newBlock(BLOCK_CAPACITY);
+ int half = block.size() / 2;
+ int from = block.start + half;
+ int count = block.end - from;
+ System.arraycopy(block.entryIds, from, upper.entryIds, 0, count);
+ System.arraycopy(block.values, from, upper.values, 0, count);
+ upper.end = count;
+ block.end = from;
+ addBlock(blockIndex + 1, upper);
+ return upper;
+ }
+
+ private void removeAt(int blockIndex, Block block, int index) {
+ block.removeAt(index);
+ size--;
+ modCount++;
+ if (block.size() == 0) {
+ // Removing the first block shifts the references to the others and the first entry IDs: draining N
+ // entries in order moves about (N / BLOCK_CAPACITY)^2 references in all. That is a small part of the
+ // entries' own handling up to about a million pending acks of one consumer in one ledger, and the AVL
+ // tree's cost at about 8 million, far above the default maxUnackedMessagesPerConsumer of 50,000
+ removeBlock(blockIndex);
+ } else {
+ firstEntryIds[blockIndex] = block.firstEntryId();
+ if (block.size() < MERGE_THRESHOLD) {
+ mergeWithNeighbor(blockIndex, block);
+ }
+ }
+ if (blocks.size() == 1) {
+ // a lone block shrinks with its entries, and the map keeps no spare block of the full capacity for it
+ spareBlock = null;
+ Block lone = blocks.get(0);
+ if (lone.capacity() > INITIAL_BLOCK_CAPACITY && lone.size() < lone.capacity() / 4) {
+ lone.resize(lone.capacity() / 2);
+ }
+ } else if (blocks.isEmpty() && (spareBlock == null || spareBlock.capacity() != INITIAL_BLOCK_CAPACITY)) {
+ // an empty map keeps a spare block of the smallest capacity only
+ spareBlock = null;
+ }
+ if (!blocks.isEmpty() && blocks.size() * 4 < peakBlocks) {
+ // the list keeps the capacity of its largest size: release it once a quarter of it is in use
+ blocks.trimToSize();
+ firstEntryIds = Arrays.copyOf(firstEntryIds, Math.max(blocks.size(), 1));
+ peakBlocks = blocks.size();
+ }
+ }
+
+ /** Merges a sparse block with the next or the previous one, when they fit in one block together. */
+ private void mergeWithNeighbor(int blockIndex, Block block) {
+ if (blockIndex + 1 < blocks.size() && block.size() + blocks.get(blockIndex + 1).size() <= BLOCK_CAPACITY / 2) {
+ moveInto(block, blocks.get(blockIndex + 1));
+ removeBlock(blockIndex + 1);
+ } else if (blockIndex > 0 && block.size() + blocks.get(blockIndex - 1).size() <= BLOCK_CAPACITY / 2) {
+ moveInto(blocks.get(blockIndex - 1), block);
+ removeBlock(blockIndex);
+ }
+ }
+
+ /** Appends the entries of the following block to the target block. */
+ private static void moveInto(Block target, Block following) {
+ int count = target.size() + following.size();
+ if (count > target.capacity()) {
+ target.grow(count);
+ } else {
+ target.compact();
+ }
+ System.arraycopy(following.entryIds, following.start, target.entryIds, target.end, following.size());
+ System.arraycopy(following.values, following.start, target.values, target.end, following.size());
+ target.end += following.size();
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
index 80345266db7e..4fc842a587f0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
@@ -18,9 +18,6 @@
*/
package org.apache.pulsar.broker.service;
-import it.unimi.dsi.fastutil.longs.Long2LongAVLTreeMap;
-import it.unimi.dsi.fastutil.longs.Long2LongMap;
-import it.unimi.dsi.fastutil.longs.Long2LongSortedMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectRBTreeMap;
import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
@@ -107,13 +104,19 @@ public class PendingAcksMap {
* Pending acknowledgments grouped by ledger id. The outer map key is ledgerId and each inner map key is entryId.
* The inner map value is packed by {@link PendingAckValues} with the remaining unacked count and sticky key hash.
*/
- private final Long2ObjectSortedMap<Long2LongSortedMap> pendingAcks;
+ private final Long2ObjectSortedMap<PendingAcksLedgerMap> pendingAcks;
private final Supplier<PendingAcksAddHandler> pendingAcksAddHandlerSupplier;
private final Supplier<PendingAcksRemoveHandler> pendingAcksRemoveHandlerSupplier;
private final Lock readLock;
private final Lock writeLock;
private boolean closed = false;
private volatile long size;
+ // A ledger map that became empty, reused for the next ledger: a consumer whose pending acks drop to none between
+ // messages, as on a topic with a low rate, doesn't allocate a ledger map for each message. Guarded by the lock
+ private PendingAcksLedgerMap emptyLedgerMap;
+ // The number of traversals whose callbacks may change the map. While one runs, a ledger map that a callback empties
+ // isn't reused: the traversal may still be iterating it. Guarded by the write lock
+ private int mutableTraversals;
PendingAcksMap(Consumer consumer, Supplier<PendingAcksAddHandler> pendingAcksAddHandlerSupplier,
Supplier<PendingAcksRemoveHandler> pendingAcksRemoveHandlerSupplier) {
@@ -154,8 +157,8 @@ public class PendingAcksMap {
&& !pendingAcksAddHandler.handleAdding(consumer, ledgerId, entryId, stickyKeyHash)) {
return false;
}
- Long2LongSortedMap ledgerPendingAcks =
- pendingAcks.computeIfAbsent(ledgerId, k -> newLedgerPendingAcks());
+ PendingAcksLedgerMap ledgerPendingAcks =
+ pendingAcks.computeIfAbsent(ledgerId, k -> takeEmptyLedgerMap());
long packedValue = PendingAckValues.pack(remainingUnacked, stickyKeyHash);
long previous = ledgerPendingAcks.put(entryId, packedValue);
if (PendingAckValues.isNotFound(previous)) {
@@ -190,15 +193,11 @@ public class PendingAcksMap {
}
}
- // this code uses for loops intentionally, don't refactor to use forEach
private void processPendingAcks(PendingAcksConsumer processor) {
- for (Long2ObjectMap.Entry<Long2LongSortedMap> entry : pendingAcks.long2ObjectEntrySet()) {
+ for (Long2ObjectMap.Entry<PendingAcksLedgerMap> entry : pendingAcks.long2ObjectEntrySet()) {
long ledgerId = entry.getLongKey();
- for (Long2LongMap.Entry pendingAckEntry : entry.getValue().long2LongEntrySet()) {
- long packedValue = pendingAckEntry.getLongValue();
- processor.accept(ledgerId, pendingAckEntry.getLongKey(),
- PendingAckValues.remainingUnacked(packedValue), PendingAckValues.stickyKeyHash(packedValue));
- }
+ entry.getValue().forEach((entryId, packedValue) -> processor.accept(ledgerId, entryId,
+ PendingAckValues.remainingUnacked(packedValue), PendingAckValues.stickyKeyHash(packedValue)));
}
}
@@ -226,6 +225,7 @@ public class PendingAcksMap {
private void internalForEachAndClear(PendingAcksConsumer processor, boolean close) {
try {
writeLock.lock();
+ mutableTraversals++;
if (close) {
closed = true;
}
@@ -246,6 +246,7 @@ public class PendingAcksMap {
pendingAcks.clear();
size = 0;
} finally {
+ mutableTraversals--;
writeLock.unlock();
}
}
@@ -260,7 +261,7 @@ public class PendingAcksMap {
public boolean contains(long ledgerId, long entryId) {
try {
readLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (ledgerMap == null) {
return false;
}
@@ -278,7 +279,7 @@ public class PendingAcksMap {
int getRemainingUnacked(long ledgerId, long entryId) {
try {
readLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (ledgerMap == null) {
return PENDING_ACK_NOT_FOUND;
}
@@ -302,7 +303,7 @@ public class PendingAcksMap {
public boolean remove(long ledgerId, long entryId, int batchSize, int stickyKeyHash) {
try {
writeLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (batchSize < 0) {
return false;
}
@@ -315,6 +316,7 @@ public class PendingAcksMap {
handleRemovePendingAck(ledgerId, entryId, stickyKeyHash);
if (ledgerMap.isEmpty()) {
pendingAcks.remove(ledgerId);
+ recycleEmptyLedgerMap(ledgerMap);
}
return true;
} finally {
@@ -334,7 +336,7 @@ public class PendingAcksMap {
public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) {
try {
writeLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (ledgerMap == null) {
return false;
}
@@ -363,7 +365,7 @@ public class PendingAcksMap {
public boolean remove(long ledgerId, long entryId) {
try {
writeLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (ledgerMap == null) {
return false;
}
@@ -375,6 +377,7 @@ public class PendingAcksMap {
handleRemovePendingAck(ledgerId, entryId, PendingAckValues.stickyKeyHash(removedEntry));
if (ledgerMap.isEmpty()) {
pendingAcks.remove(ledgerId);
+ recycleEmptyLedgerMap(ledgerMap);
}
return true;
} finally {
@@ -390,7 +393,7 @@ public class PendingAcksMap {
int removeAndGetRemainingUnacked(long ledgerId, long entryId) {
try {
writeLock.lock();
- Long2LongSortedMap ledgerMap = pendingAcks.get(ledgerId);
+ PendingAcksLedgerMap ledgerMap = pendingAcks.get(ledgerId);
if (ledgerMap == null) {
return PENDING_ACK_NOT_FOUND;
}
@@ -402,6 +405,7 @@ public class PendingAcksMap {
handleRemovePendingAck(ledgerId, entryId, PendingAckValues.stickyKeyHash(removedEntry));
if (ledgerMap.isEmpty()) {
pendingAcks.remove(ledgerId);
+ recycleEmptyLedgerMap(ledgerMap);
}
return PendingAckValues.remainingUnacked(removedEntry);
} finally {
@@ -447,31 +451,31 @@ public class PendingAcksMap {
if (useWriteLock) {
writeLock.lock();
acquiredWriteLock = true;
+ mutableTraversals++;
} else {
readLock.lock();
}
- ObjectBidirectionalIterator<Long2ObjectMap.Entry<Long2LongSortedMap>> ledgerMapIterator =
+ ObjectBidirectionalIterator<Long2ObjectMap.Entry<PendingAcksLedgerMap>> ledgerMapIterator =
pendingAcks.headMap(markDeleteLedgerId + 1).long2ObjectEntrySet().iterator();
while (ledgerMapIterator.hasNext()) {
- Long2ObjectMap.Entry<Long2LongSortedMap> entry = ledgerMapIterator.next();
+ Long2ObjectMap.Entry<PendingAcksLedgerMap> entry = ledgerMapIterator.next();
long ledgerId = entry.getLongKey();
- Long2LongSortedMap ledgerMap = entry.getValue();
- Long2LongSortedMap ledgerMapHead;
- if (ledgerId == markDeleteLedgerId) {
- ledgerMapHead = ledgerMap.headMap(markDeleteEntryId + 1);
- } else {
- ledgerMapHead = ledgerMap;
- }
- ObjectBidirectionalIterator<Long2LongMap.Entry> entryMapIterator =
- ledgerMapHead.long2LongEntrySet().iterator();
- while (entryMapIterator.hasNext()) {
- Long2LongMap.Entry pendingAckEntry = entryMapIterator.next();
- long entryId = pendingAckEntry.getLongKey();
- if (!acquiredWriteLock) {
+ PendingAcksLedgerMap ledgerMap = entry.getValue();
+ long maxEntryId = ledgerId == markDeleteLedgerId ? markDeleteEntryId : Long.MAX_VALUE;
+ if (!acquiredWriteLock) {
+ if (ledgerMap.hasEntriesUpTo(maxEntryId) || ledgerMap.isEmpty()) {
retryWithWriteLock = true;
return;
}
- long packedValue = pendingAckEntry.getLongValue();
+ continue;
+ }
+ // One entry at a time, the callbacks before its removal, as an iterator would. The callbacks may add
+ // and remove entries: the next entry is the first one after the removed one, so an entry added before
+ // it isn't removed, as with the tree map's iterator before, and the removal ends
+ long entryId = ledgerMap.hasEntriesUpTo(maxEntryId) ? ledgerMap.firstEntryId()
+ : PendingAcksLedgerMap.NO_ENTRY;
+ while (entryId != PendingAcksLedgerMap.NO_ENTRY && entryId <= maxEntryId) {
+ long packedValue = ledgerMap.get(entryId);
int batchSize = PendingAckValues.remainingUnacked(packedValue);
int stickyKeyHash = PendingAckValues.stickyKeyHash(packedValue);
if (pendingAcksRemoveHandler != null) {
@@ -484,15 +488,14 @@ public class PendingAcksMap {
if (removedEntryCallback != null) {
removedEntryCallback.accept(ledgerId, entryId, batchSize, stickyKeyHash);
}
- entryMapIterator.remove();
- size--;
+ if (!PendingAckValues.isNotFound(ledgerMap.remove(entryId))) {
+ size--;
+ }
+ entryId = ledgerMap.higherEntryId(entryId);
}
if (ledgerMap.isEmpty()) {
- if (!acquiredWriteLock) {
- retryWithWriteLock = true;
- return;
- }
ledgerMapIterator.remove();
+ recycleEmptyLedgerMap(ledgerMap);
}
}
} finally {
@@ -500,6 +503,7 @@ public class PendingAcksMap {
pendingAcksRemoveHandler.endBatch();
}
if (acquiredWriteLock) {
+ mutableTraversals--;
writeLock.unlock();
} else {
readLock.unlock();
@@ -510,10 +514,19 @@ public class PendingAcksMap {
}
}
- private static Long2LongSortedMap newLedgerPendingAcks() {
- Long2LongAVLTreeMap ledgerPendingAcks = new Long2LongAVLTreeMap();
- ledgerPendingAcks.defaultReturnValue(PendingAckValues.PACKED_NOT_FOUND);
- return ledgerPendingAcks;
+ private void recycleEmptyLedgerMap(PendingAcksLedgerMap ledgerMap) {
+ if (mutableTraversals == 0) {
+ emptyLedgerMap = ledgerMap;
+ }
+ }
+
+ private PendingAcksLedgerMap takeEmptyLedgerMap() {
+ PendingAcksLedgerMap ledgerMap = emptyLedgerMap;
+ if (ledgerMap == null) {
+ return new PendingAcksLedgerMap();
+ }
+ emptyLedgerMap = null;
+ return ledgerMap;
}
private void handleRemovePendingAck(long ledgerId, long entryId, int stickyKeyHash) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapTest.java
new file mode 100644
index 000000000000..bd73c9a5c0cb
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksLedgerMapTest.java
@@ -0,0 +1,501 @@
+/*
+ * 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.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.TreeMap;
+import org.testng.annotations.Test;
+
+public class PendingAcksLedgerMapTest {
+
+ @Test
+ public void keepsEntriesInOrderWithAppendsInsertsAndRemovals() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ assertThat(map.put(5, value(1))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.put(7, value(2))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ // out of order, as a redelivery adds it
+ assertThat(map.put(6, value(3))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.put(1, value(4))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.put(6, value(5))).isEqualTo(value(3));
+ assertThat(entries(map)).containsExactly(1L, 5L, 6L, 7L);
+ assertThat(map.get(6)).isEqualTo(value(5));
+ assertThat(map.get(2)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+
+ assertThat(map.remove(5)).isEqualTo(value(1));
+ assertThat(map.remove(5)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.containsKey(5)).isFalse();
+ // a removed entry can come back
+ assertThat(map.put(5, value(6))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(entries(map)).containsExactly(1L, 5L, 6L, 7L);
+
+ List<Long> removed = new ArrayList<>();
+ assertThat(map.hasEntriesUpTo(0)).isFalse();
+ assertThat(map.hasEntriesUpTo(6)).isTrue();
+ while (map.hasEntriesUpTo(6)) {
+ removed.add(map.firstEntryId());
+ map.removeFirst();
+ }
+ assertThat(removed).containsExactly(1L, 5L, 6L);
+ assertThat(entries(map)).containsExactly(7L);
+ assertThat(map.size()).isEqualTo(1);
+ assertThat(map.remove(7)).isEqualTo(value(2));
+ assertThat(map.isEmpty()).isTrue();
+ assertThat(map.hasEntriesUpTo(Long.MAX_VALUE)).isFalse();
+ }
+
+ @Test
+ public void findsAndRemovesEntriesAckedInOrder() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int count = PendingAcksLedgerMap.BLOCK_CAPACITY * 3 + 5;
+ for (int i = 0; i < count; i++) {
+ map.put(i * 2, value(i));
+ }
+ // acked in the order they were sent: each is the first entry, across the blocks
+ for (int i = 0; i < count; i++) {
+ assertThat(map.get(i * 2)).isEqualTo(value(i));
+ // not an entry, just after the first
+ assertThat(map.get(i * 2 + 1)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.remove(i * 2)).isEqualTo(value(i));
+ assertThat(map.get(i * 2)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.size()).isEqualTo(count - i - 1);
+ }
+ assertThat(map.isEmpty()).isTrue();
+ assertThat(map.remove(0)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.get(0)).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ }
+
+ @Test
+ public void behavesAsASortedMap() {
+ // random operations in the patterns of dispatch and acks: mostly increasing entry IDs, some redeliveries,
+ // removals near the start and range removals up to a mark-delete position
+ Random random = new Random(1);
+ for (int round = 0; round < 50; round++) {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ TreeMap<Long, Long> reference = new TreeMap<>();
+ long nextEntryId = 0;
+ for (int step = 0; step < 5000; step++) {
+ int operation = random.nextInt(100);
+ if (operation < 45) {
+ long entryId = random.nextInt(10) == 0 && nextEntryId > 0
+ ? random.nextLong(nextEntryId) : nextEntryId++;
+ long v = value(random.nextInt(1000));
+ Long previous = reference.put(entryId, v);
+ assertThat(map.put(entryId, v))
+ .isEqualTo(previous != null ? previous : PendingAckValues.PACKED_NOT_FOUND);
+ } else if (operation < 90) {
+ long entryId = reference.isEmpty() || random.nextInt(4) == 0
+ ? random.nextLong(nextEntryId + 1) : nthKey(reference, random.nextInt(
+ Math.min(reference.size(), 20)));
+ Long previous = reference.remove(entryId);
+ assertThat(map.remove(entryId))
+ .isEqualTo(previous != null ? previous : PendingAckValues.PACKED_NOT_FOUND);
+ } else if (operation < 95) {
+ long upTo = reference.isEmpty() ? nextEntryId : reference.firstKey() + random.nextInt(30);
+ Map<Long, Long> head = new TreeMap<>(reference.headMap(upTo, true));
+ List<Long> removedIds = new ArrayList<>();
+ List<Long> removedValues = new ArrayList<>();
+ assertThat(map.hasEntriesUpTo(upTo)).isEqualTo(!head.isEmpty());
+ while (map.hasEntriesUpTo(upTo)) {
+ removedIds.add(map.firstEntryId());
+ removedValues.add(map.firstValue());
+ map.removeFirst();
+ }
+ assertThat(removedIds).containsExactlyElementsOf(head.keySet());
+ assertThat(removedValues).containsExactlyElementsOf(head.values());
+ reference.headMap(upTo, true).clear();
+ } else {
+ long entryId = random.nextLong(nextEntryId + 1);
+ Long expected = reference.get(entryId);
+ assertThat(map.get(entryId))
+ .isEqualTo(expected != null ? expected : PendingAckValues.PACKED_NOT_FOUND);
+ assertThat(map.containsKey(entryId)).isEqualTo(expected != null);
+ }
+ assertThat(map.size()).isEqualTo(reference.size());
+ map.checkInvariants();
+ if (step % 97 == 0) {
+ assertThat(contents(map)).isEqualTo(reference);
+ }
+ }
+ List<Long> ids = new ArrayList<>();
+ List<Long> values = new ArrayList<>();
+ map.forEach((entryId, v) -> {
+ ids.add(entryId);
+ values.add(v);
+ });
+ assertThat(ids).containsExactlyElementsOf(reference.keySet());
+ assertThat(values).containsExactlyElementsOf(reference.values());
+ }
+ }
+
+ @Test(timeOut = 30_000)
+ public void reinsertsARemovedPrefixInAscendingOrderWithoutCopyingTheRest() {
+ // explicit redelivery removes a consumer's pending acks and dispatches them again, in the order of their
+ // positions: re-adding them mustn't shift every following entry each time
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int count = 200_000;
+ for (int entryId = 0; entryId < count; entryId++) {
+ map.put(entryId, value(1));
+ }
+ for (int entryId = 0; entryId < count / 2; entryId++) {
+ assertThat(map.remove(entryId)).isEqualTo(value(1));
+ }
+ for (int entryId = 0; entryId < count / 2; entryId++) {
+ assertThat(map.put(entryId, value(2))).isEqualTo(PendingAckValues.PACKED_NOT_FOUND);
+ }
+ assertThat(map.size()).isEqualTo(count);
+ long[] expected = {0};
+ map.forEach((entryId, v) -> {
+ assertThat(entryId).isEqualTo(expected[0]++);
+ assertThat(v).isEqualTo(entryId < count / 2 ? value(2) : value(1));
+ });
+ }
+
+ @Test
+ public void releasesTheMemoryOfRemovedEntries() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int count = 64 * PendingAcksLedgerMap.BLOCK_CAPACITY;
+ for (int entryId = 0; entryId < count; entryId++) {
+ map.put(entryId, value(1));
+ }
+ assertThat(map.blockCount()).isEqualTo(64);
+ // keep every 64th entry: the sparse blocks merge with their neighbors
+ for (int entryId = 0; entryId < count; entryId++) {
+ if (entryId % 64 != 0) {
+ map.remove(entryId);
+ }
+ }
+ // 128 entries, in blocks of at most half a block after the merges
+ assertThat(map.size()).isEqualTo(128);
+ assertThat(map.blockCount()).isLessThanOrEqualTo(4);
+ for (int entryId = 0; entryId < count; entryId += 64) {
+ assertThat(map.get(entryId)).isEqualTo(value(1));
+ }
+ while (!map.isEmpty()) {
+ map.removeFirst();
+ }
+ assertThat(map.blockCount()).isZero();
+ }
+
+ @Test
+ public void splitsAFullBlockForAnEntryInsertedInIt() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ for (int entryId = 0; entryId < PendingAcksLedgerMap.BLOCK_CAPACITY; entryId++) {
+ map.put(entryId * 2L, value(1));
+ }
+ assertThat(map.blockCount()).isEqualTo(1);
+ map.put(1, value(2));
+ map.put(2L * PendingAcksLedgerMap.BLOCK_CAPACITY - 3, value(3));
+ assertThat(map.blockCount()).isEqualTo(2);
+ assertThat(map.get(1)).isEqualTo(value(2));
+ assertThat(map.get(2L * PendingAcksLedgerMap.BLOCK_CAPACITY - 3)).isEqualTo(value(3));
+ assertThat(map.size()).isEqualTo(PendingAcksLedgerMap.BLOCK_CAPACITY + 2);
+ List<Long> ids = entries(map);
+ assertThat(ids).isSorted().doesNotHaveDuplicates();
+ }
+
+ @Test
+ public void insertsIntoTheUpperHalfOfAFullBlock() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ for (int entryId = 0; entryId < PendingAcksLedgerMap.BLOCK_CAPACITY; entryId++) {
+ map.put(entryId * 2L, value(1));
+ }
+ // after the first block's middle, so that it goes into the new upper block of the split
+ long entryId = 2L * PendingAcksLedgerMap.BLOCK_CAPACITY - 5;
+ map.put(entryId, value(2));
+ assertThat(map.blockCount()).isEqualTo(2);
+ assertThat(map.get(entryId)).isEqualTo(value(2));
+ assertThat(entries(map)).isSorted().doesNotHaveDuplicates().hasSize(PendingAcksLedgerMap.BLOCK_CAPACITY + 1);
+ }
+
+ @Test
+ public void forEachContinuesAfterTheCurrentEntryWhenTheConsumerChangesTheMap() {
+ // The consumer removes the current entry, removes or adds other entries before and after it, in blocks that
+ // shift and split. The traversal visits the entries after the current one as they are after the consumer's
+ // changes, as TreeMap.higherKey finds them
+ for (int round = 0; round < 20; round++) {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ TreeMap<Long, Long> reference = new TreeMap<>();
+ for (long entryId = 0; entryId < 1000; entryId += 2) {
+ map.put(entryId, value(1));
+ reference.put(entryId, value(1));
+ }
+ long seed = round;
+ List<Long> visited = new ArrayList<>();
+ map.forEach((entryId, v) -> {
+ visited.add(entryId);
+ changeAt(entryId, seed, (id, add) -> {
+ if (add) {
+ map.put(id, value(1));
+ } else {
+ map.remove(id);
+ }
+ });
+ });
+
+ List<Long> expected = new ArrayList<>();
+ for (Long entryId = reference.firstKey(); entryId != null; entryId = reference.higherKey(entryId)) {
+ expected.add(entryId);
+ changeAt(entryId, seed, (id, add) -> {
+ if (add) {
+ reference.put(id, value(1));
+ } else {
+ reference.remove(id);
+ }
+ });
+ }
+ assertThat(visited).as("round %d", round).isEqualTo(expected);
+ assertThat(contents(map)).as("round %d", round).isEqualTo(reference);
+ map.checkInvariants();
+ }
+ }
+
+ @Test
+ public void forEachVisitsEveryEntryWhileTheConsumerDrainsTheMap() {
+ // Removing the current entry, and every other one after it, empties blocks, which are removed, and leaves
+ // sparse blocks, which merge: the traversal still visits each remaining entry once, in order
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int count = 5 * PendingAcksLedgerMap.BLOCK_CAPACITY;
+ for (long entryId = 0; entryId < count; entryId++) {
+ map.put(entryId, value(1));
+ }
+ int blocksBefore = map.blockCount();
+ List<Long> visited = new ArrayList<>();
+ int[] minBlocks = {blocksBefore};
+ map.forEach((entryId, v) -> {
+ visited.add(entryId);
+ map.remove(entryId);
+ if (entryId % 64 == 0) {
+ // thin out the entries ahead, which leaves sparse blocks that merge
+ for (long ahead = entryId + 2; ahead < entryId + 64; ahead += 2) {
+ map.remove(ahead);
+ }
+ }
+ minBlocks[0] = Math.min(minBlocks[0], map.blockCount());
+ });
+
+ List<Long> expected = new ArrayList<>();
+ for (long entryId = 0; entryId < count; entryId++) {
+ if (entryId % 64 == 0 || entryId % 2 == 1) {
+ expected.add(entryId);
+ }
+ }
+ assertThat(visited).isEqualTo(expected);
+ assertThat(map.isEmpty()).isTrue();
+ assertThat(map.blockCount()).isZero();
+ assertThat(minBlocks[0]).isLessThan(blocksBefore);
+ map.checkInvariants();
+ }
+
+ @Test
+ public void startsWithASmallFirstBlockThatGrows() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ map.put(0, value(1));
+ assertThat(map.blockCapacity(0)).isEqualTo(PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ for (long entryId = 1; entryId <= PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY; entryId++) {
+ map.put(entryId, value(1));
+ }
+ assertThat(map.blockCapacity(0)).isEqualTo(2 * PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ for (long entryId = PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY + 1;
+ entryId < PendingAcksLedgerMap.BLOCK_CAPACITY; entryId++) {
+ map.put(entryId, value(1));
+ }
+ // one block of the full capacity, and then a new one of the full capacity
+ assertThat(map.blockCount()).isEqualTo(1);
+ assertThat(map.blockCapacity(0)).isEqualTo(PendingAcksLedgerMap.BLOCK_CAPACITY);
+ map.put(PendingAcksLedgerMap.BLOCK_CAPACITY, value(1));
+ assertThat(map.blockCount()).isEqualTo(2);
+ assertThat(map.blockCapacity(1)).isEqualTo(PendingAcksLedgerMap.BLOCK_CAPACITY);
+ map.checkInvariants();
+ }
+
+ @Test
+ public void growsASmallFullBlockForAnEntryInsertedInIt() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ for (long entryId = 0; entryId < 2L * PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY; entryId += 2) {
+ map.put(entryId, value(1));
+ }
+ assertThat(map.blockCapacity(0)).isEqualTo(PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ // before the first entry, and between two entries
+ map.put(-1, value(2));
+ map.put(5, value(3));
+ assertThat(map.blockCount()).isEqualTo(1);
+ assertThat(map.blockCapacity(0)).isEqualTo(2 * PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ assertThat(map.firstEntryId()).isEqualTo(-1);
+ assertThat(map.get(-1)).isEqualTo(value(2));
+ assertThat(map.get(5)).isEqualTo(value(3));
+ assertThat(entries(map)).isSorted().doesNotHaveDuplicates()
+ .hasSize(PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY + 2);
+ map.checkInvariants();
+ }
+
+ @Test
+ public void reusesTheLastEmptiedBlockForAWindowThatMoves() {
+ // a consumer's window of pending acks moving through a ledger: acks remove the oldest entries and dispatches
+ // append new ones
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int window = 4 * PendingAcksLedgerMap.BLOCK_CAPACITY;
+ for (long entryId = 0; entryId < window; entryId++) {
+ map.put(entryId, value(1));
+ }
+ int allocated = map.allocatedBlocks();
+ for (long entryId = 0; entryId < 100L * PendingAcksLedgerMap.BLOCK_CAPACITY; entryId++) {
+ assertThat(map.remove(entryId)).isEqualTo(value(1));
+ map.put(entryId + window, value(1));
+ }
+ // a block more than the window's, while the oldest block still has entries when a new one starts
+ assertThat(map.allocatedBlocks()).isLessThanOrEqualTo(allocated + 1);
+ assertThat(map.size()).isEqualTo(window);
+ map.checkInvariants();
+ // an empty map keeps a spare block of the smallest capacity only, which it starts again with
+ while (!map.isEmpty()) {
+ map.removeFirst();
+ map.checkInvariants();
+ }
+ int allocatedWhenEmpty = map.allocatedBlocks();
+ map.put(0, value(1));
+ assertThat(map.blockCapacity(0)).isEqualTo(PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ assertThat(map.allocatedBlocks()).isEqualTo(allocatedWhenEmpty);
+ map.checkInvariants();
+ }
+
+ @Test
+ public void shrinksALoneBlockAsItsEntriesAreRemoved() {
+ // a map that held many entries and now holds a few keeps a small block, and no spare one
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ int count = 5 * PendingAcksLedgerMap.BLOCK_CAPACITY;
+ for (long entryId = 0; entryId < count; entryId++) {
+ map.put(entryId, value(1));
+ }
+ // ack holes: every 100th entry stays pending
+ for (long entryId = 0; entryId < count; entryId++) {
+ if (entryId % 100 != 0) {
+ map.remove(entryId);
+ map.checkInvariants();
+ }
+ }
+ assertThat(map.size()).isEqualTo(7);
+ assertThat(map.blockCount()).isEqualTo(1);
+ assertThat(map.blockCapacity(0)).isEqualTo(PendingAcksLedgerMap.INITIAL_BLOCK_CAPACITY);
+ assertThat(entries(map)).containsExactly(0L, 100L, 200L, 300L, 400L, 500L, 600L);
+ }
+
+ @Test
+ public void staysASortedMapWithRemovalsAcrossTheWindowAndAppendsAfterThem() {
+ // Removals anywhere in a large window empty, merge and shrink blocks, whose arrays are reused as spare
+ // blocks for the appends, inserts and splits that follow
+ Random random = new Random(2);
+ for (int round = 0; round < 20; round++) {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ TreeMap<Long, Long> reference = new TreeMap<>();
+ long nextEntryId = 0;
+ for (; nextEntryId < 1000; nextEntryId++) {
+ map.put(nextEntryId, value(1));
+ reference.put(nextEntryId, value(1));
+ }
+ for (int step = 0; step < 4000; step++) {
+ int operation = random.nextInt(10);
+ if (operation < 5 && !reference.isEmpty()) {
+ // a removal anywhere in the window
+ long entryId = random.nextLong(nextEntryId);
+ Long previous = reference.remove(entryId);
+ assertThat(map.remove(entryId))
+ .isEqualTo(previous != null ? previous : PendingAckValues.PACKED_NOT_FOUND);
+ } else if (operation < 8) {
+ long v = value(1 + random.nextInt(100));
+ reference.put(nextEntryId, v);
+ map.put(nextEntryId++, v);
+ } else {
+ // a redelivery, out of order
+ long entryId = random.nextLong(nextEntryId);
+ long v = value(1 + random.nextInt(100));
+ Long previous = reference.put(entryId, v);
+ assertThat(map.put(entryId, v))
+ .isEqualTo(previous != null ? previous : PendingAckValues.PACKED_NOT_FOUND);
+ }
+ map.checkInvariants();
+ assertThat(map.size()).isEqualTo(reference.size());
+ }
+ assertThat(contents(map)).as("round %d", round).isEqualTo(reference);
+ }
+ }
+
+ @Test
+ public void higherEntryIdFindsTheNextEntryAcrossBlocks() {
+ PendingAcksLedgerMap map = new PendingAcksLedgerMap();
+ assertThat(map.higherEntryId(0)).isEqualTo(PendingAcksLedgerMap.NO_ENTRY);
+ for (long entryId = 10; entryId < 10 + 3 * PendingAcksLedgerMap.BLOCK_CAPACITY; entryId += 2) {
+ map.put(entryId, value(1));
+ }
+ assertThat(map.blockCount()).isGreaterThan(1);
+ long last = 10 + 3 * PendingAcksLedgerMap.BLOCK_CAPACITY - 2;
+ assertThat(map.higherEntryId(Long.MIN_VALUE + 1)).isEqualTo(10);
+ assertThat(map.higherEntryId(10)).isEqualTo(12);
+ assertThat(map.higherEntryId(11)).isEqualTo(12);
+ for (long entryId = 10; entryId < last; entryId++) {
+ assertThat(map.higherEntryId(entryId)).as("after %d", entryId).isEqualTo(entryId + 2 - entryId % 2);
+ }
+ assertThat(map.higherEntryId(last)).isEqualTo(PendingAcksLedgerMap.NO_ENTRY);
+ assertThat(map.higherEntryId(Long.MAX_VALUE)).isEqualTo(PendingAcksLedgerMap.NO_ENTRY);
+ }
+
+ private interface Change {
+ void apply(long entryId, boolean add);
+ }
+
+ // The changes that a consumer makes when it visits the entry, the same for the map and the reference
+ private static void changeAt(long entryId, long seed, Change change) {
+ Random random = new Random(seed * 1_000_003 + entryId);
+ if (random.nextInt(3) == 0) {
+ change.apply(entryId, false);
+ }
+ for (int i = random.nextInt(4); i > 0; i--) {
+ change.apply(random.nextInt(1200), random.nextBoolean());
+ }
+ }
+
+ private static TreeMap<Long, Long> contents(PendingAcksLedgerMap map) {
+ TreeMap<Long, Long> contents = new TreeMap<>();
+ map.forEach(contents::put);
+ return contents;
+ }
+
+ private static long value(int remainingUnacked) {
+ return PendingAckValues.pack(remainingUnacked, remainingUnacked * 31);
+ }
+
+ private static long nthKey(TreeMap<Long, Long> map, int n) {
+ int i = 0;
+ for (Long key : map.keySet()) {
+ if (i++ == n) {
+ return key;
+ }
+ }
+ throw new IllegalArgumentException();
+ }
+
+ private static List<Long> entries(PendingAcksLedgerMap map) {
+ List<Long> ids = new ArrayList<>();
+ map.forEach((entryId, v) -> ids.add(entryId));
+ return ids;
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
index 59aa302ebb15..ae8b432fb31f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
@@ -131,6 +131,185 @@ public class PendingAcksMapTest {
assertEquals(pendingAcksMap.size(), 1);
}
+ @Test
+ public void forEachAndClear_VisitsTheEntriesAfterOneThatTheCallbackRemoves() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 3L, 1, 125);
+ List<Long> visited = new ArrayList<>();
+
+ pendingAcksMap.forEachAndClear((ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ visited.add(entryId);
+ if (entryId == 2L) {
+ pendingAcksMap.remove(1L, 2L);
+ }
+ });
+
+ assertEquals(visited, List.of(1L, 2L, 3L));
+ assertEquals(pendingAcksMap.size(), 0);
+ }
+
+ @Test(timeOut = 10_000)
+ public void removeAllUpTo_DoesntRemoveAnEntryThatTheCallbackAddsBeforeTheCurrentOne() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 3L, 1, 125);
+ List<Long> removed = new ArrayList<>();
+
+ // adding an entry before the current one, and one after it, which is removed too
+ pendingAcksMap.removeAllUpTo(1L, 3L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ removed.add(entryId);
+ if (entryId == 1L) {
+ pendingAcksMap.addPendingAckIfAllowed(1L, 0L, 1, 122);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
+ } else if (entryId == 2L) {
+ // re-adding the removed first entry, which would be removed again without an end
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ }
+ });
+
+ assertEquals(removed, List.of(1L, 2L, 3L));
+ assertTrue(pendingAcksMap.contains(1L, 0L));
+ assertTrue(pendingAcksMap.contains(1L, 1L));
+ assertEquals(pendingAcksMap.size(), 2);
+ }
+
+ @Test
+ public void removeAllUpTo_KeepsTheSizeWhenTheCallbackRemovesTheCurrentEntry() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 3L, 1, 125);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 4L, 1, 126);
+
+ pendingAcksMap.removeAllUpTo(1L, 2L, (ledgerId, entryId, batchSize, stickyKeyHash) ->
+ pendingAcksMap.remove(ledgerId, entryId));
+
+ // each removed entry is counted once
+ assertEquals(pendingAcksMap.size(), 2);
+ assertTrue(pendingAcksMap.contains(1L, 3L));
+ assertTrue(pendingAcksMap.contains(1L, 4L));
+ }
+
+ @Test
+ public void forEachAndClear_DoesntVisitAnEntryThatTheCallbackRemoves() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ for (long entryId = 1; entryId <= 5; entryId++) {
+ pendingAcksMap.addPendingAckIfAllowed(1L, entryId, 1, 120 + (int) entryId);
+ }
+ List<Long> visited = new ArrayList<>();
+
+ pendingAcksMap.forEachAndClear((ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ visited.add(entryId);
+ if (entryId == 2L) {
+ pendingAcksMap.remove(1L, 3L);
+ }
+ });
+
+ assertEquals(visited, List.of(1L, 2L, 4L, 5L));
+ assertEquals(pendingAcksMap.size(), 0);
+ }
+
+ @Test
+ public void reusesAnEmptiedLedgerMapForTheNextEntries() {
+ // a consumer whose pending acks drop to none between messages
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ for (long entryId = 0; entryId < 100; entryId++) {
+ assertTrue(pendingAcksMap.addPendingAckIfAllowed(entryId / 10, entryId, 1, 123));
+ assertEquals(pendingAcksMap.getRemainingUnacked(entryId / 10, entryId), 1);
+ assertEquals(pendingAcksMap.removeAndGetRemainingUnacked(entryId / 10, entryId), 1);
+ assertEquals(pendingAcksMap.size(), 0);
+ assertFalse(pendingAcksMap.contains(entryId / 10, entryId));
+ }
+ pendingAcksMap.addPendingAckIfAllowed(20L, 1L, 2, 123);
+ pendingAcksMap.addPendingAckIfAllowed(20L, 2L, 3, 124);
+ assertEquals(pendingAcksMap.getRemainingUnacked(20L, 1L), 2);
+ assertEquals(pendingAcksMap.getRemainingUnacked(20L, 2L), 3);
+ assertEquals(pendingAcksMap.size(), 2);
+ }
+
+ @Test
+ public void removeAllUpTo_KeepsAnEntryThatTheCallbackAddsToAnotherLedgerAfterEmptyingTheCurrentOne() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+
+ pendingAcksMap.removeAllUpTo(1L, 1L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ // emptying the ledger map that the removal iterates, then adding an entry to another ledger
+ pendingAcksMap.remove(1L, 1L);
+ pendingAcksMap.addPendingAckIfAllowed(2L, 1L, 1, 124);
+ });
+
+ assertFalse(pendingAcksMap.contains(1L, 1L));
+ assertTrue(pendingAcksMap.contains(2L, 1L));
+ assertEquals(pendingAcksMap.size(), 1);
+ }
+
+ @Test
+ public void forEachAndClear_DoesntVisitAnEntryThatTheCallbackAddsToAnotherLedgerAfterEmptyingTheCurrentOne() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ List<String> visited = new ArrayList<>();
+
+ pendingAcksMap.forEachAndClear((ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ visited.add(ledgerId + ":" + entryId);
+ if (ledgerId == 1L) {
+ pendingAcksMap.remove(1L, 1L);
+ pendingAcksMap.addPendingAckIfAllowed(2L, 5L, 1, 124);
+ }
+ });
+
+ assertEquals(visited, List.of("1:1"));
+ assertEquals(pendingAcksMap.size(), 0);
+ }
+
+ @Test
+ public void removeAllUpTo_KeepsTheSizeWhenTheCallbackThrows() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
+
+ try {
+ pendingAcksMap.removeAllUpTo(1L, 2L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ if (entryId == 2L) {
+ throw new IllegalStateException("failing callback");
+ }
+ });
+ } catch (IllegalStateException expected) {
+ // the entries before the failing one were removed, the failing one wasn't
+ }
+
+ assertFalse(pendingAcksMap.contains(1L, 1L));
+ assertTrue(pendingAcksMap.contains(1L, 2L));
+ assertEquals(pendingAcksMap.size(), 1);
+ assertTrue(pendingAcksMap.remove(1L, 2L));
+ assertEquals(pendingAcksMap.size(), 0);
+ }
+
+ @Test
+ public void removeAllUpTo_KeepsTheSizeWhenTheCallbackAddsAnEntry() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 1, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
+
+ pendingAcksMap.removeAllUpTo(1L, 1L, (ledgerId, entryId, batchSize, stickyKeyHash) ->
+ pendingAcksMap.addPendingAckIfAllowed(1L, 3L, 1, 125));
+
+ assertFalse(pendingAcksMap.contains(1L, 1L));
+ assertTrue(pendingAcksMap.contains(1L, 2L));
+ assertTrue(pendingAcksMap.contains(1L, 3L));
+ assertEquals(pendingAcksMap.size(), 2);
+ }
+
@Test
public void removeAllUpTo_RemovesAllPendingAcksUpToSpecifiedEntryAcrossMultipleLedgers() {
Consumer consumer = createMockConsumer("consumer1");