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/09/09 15:14:54 UTC
(pulsar) branch branch-4.2 updated (478da98e8f49 -> b3dcc415750c)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 478da98e8f49 [fix][client] Fix incorrect topic in OTel spans for partitioned producers (#25896)
new d144107e5d65 [improve][broker] Avoid iterator allocation in BrokerInterceptors hot paths and restore configured ordering (#26462)
new 77bab3bb4ae8 [improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)
new 52f6296d368d [fix][cli] pulsar-perf: register subcommand classes with picocli instead of instances (#26467)
new a2ddf086e3ea [fix][cli] Apply PULSAR_MEM and PULSAR_GC in bin/pulsar-perf (#26465)
new b633a2bdc773 [improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
new 2d600cfc2c15 [improve][ml] Parse message metadata at entry cache insert time instead of under the wrapper write lock (#26463)
new e6087f0a9ba4 [fix][broker] Fix NPE in ManagedLedgerInterceptorImpl when AppendIndexMetadataInterceptor isn't configured (#26497)
new c7c963cfa668 [fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
new 62721fefaf28 [fix][admin] Reject empty cluster migration URL: fix inverted ClusterUrl.isEmpty() (#26473)
new 6a5b19d640f1 [fix][client] Release the reserved memory only once when failing a send in a terminal state (#26476)
new 65ca2a6efdc6 [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps (#26495)
new 59311c967c40 [improve][cli] Default the pulsar-perf client memory limit to half of the JVM max direct memory (#26491)
new 4d59e8d33cc9 [fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
new b3dcc415750c [fix][ml] Fix active cursor retention and deferred-update synchronization (#26511)
The 14 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
bin/pulsar-perf | 20 +
managed-ledger/pom.xml | 5 +
.../bookkeeper/mledger/ManagedLedgerConfig.java | 18 +
.../impl/ActiveManagedCursorContainerImpl.java | 208 +++++++--
.../mledger/impl/cache/EntryCacheDisabled.java | 4 +-
.../bookkeeper/mledger/impl/cache/RangeCache.java | 14 +-
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 35 +-
.../mledger/impl/cache/RangeEntryCacheImpl.java | 21 +-
.../ActiveManagedCursorContainerRetentionTest.java | 491 +++++++++++++++++++++
.../mledger/impl/cache/EntryCacheDisabledTest.java | 136 ++++++
.../impl/cache/RangeEntryCacheImplTest.java | 278 +++++++++++-
.../ActiveManagedCursorContainerBenchmark.java | 44 +-
...ctiveManagedCursorContainerCohortBenchmark.java | 82 ++++
.../broker/intercept/BrokerInterceptors.java | 55 +--
.../intercept/ManagedLedgerInterceptorImpl.java | 8 +-
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../impl/AbstractSnapshotAbortedTxnProcessor.java | 186 ++++++++
.../SingleSnapshotAbortedTxnProcessorImpl.java | 44 +-
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 96 ++--
.../buffer/impl/TopicTransactionBuffer.java | 212 ++++++---
.../pendingack/impl/MLPendingAckStoreProvider.java | 2 +
.../org/apache/pulsar/broker/web/WebService.java | 2 +-
.../org/apache/pulsar/broker/admin/AdminTest.java | 18 +-
.../broker/intercept/BrokerInterceptorTest.java | 2 +-
.../intercept/ExceptionsBrokerInterceptorTest.java | 2 +-
.../ManagedLedgerInterceptorImplTest.java | 111 +++++
.../service/persistent/ReaderMessageTTLTest.java | 99 +++++
.../pulsar/broker/transaction/TransactionTest.java | 36 +-
.../AbstractSnapshotAbortedTxnProcessorTest.java | 302 +++++++++++++
...napshotSegmentAbortedTxnProcessorCloseTest.java | 173 ++++++++
.../impl/TopicTransactionBufferCloseTest.java | 276 ++++++++++++
.../impl/TopicTransactionBufferRecoveryTest.java | 88 ++++
.../common/policies/data/ClusterPolicies.java | 11 +-
.../common/policies/data/ClusterUrlTest.java | 60 +++
.../apache/pulsar/client/impl/ProducerImpl.java | 2 +-
.../pulsar/client/impl/ProducerImplTest.java | 46 ++
.../apache/pulsar/common/protocol/Commands.java | 25 ++
.../pulsar/common/protocol/CommandUtilsTest.java | 24 +
.../proxy/socket/client/PerformanceClient.java | 31 +-
.../socket/client/SimpleTestProducerSocket.java | 13 +-
.../testclient/CmdGenerateDocumentation.java | 7 +-
.../pulsar/testclient/ManagedLedgerWriter.java | 33 +-
.../apache/pulsar/testclient/PerfClientUtils.java | 17 +
.../testclient/PerformanceBaseArguments.java | 17 +-
.../pulsar/testclient/PerformanceConsumer.java | 38 +-
.../pulsar/testclient/PerformanceProducer.java | 41 +-
.../pulsar/testclient/PerformanceReader.java | 26 +-
.../pulsar/testclient/PerformanceTransaction.java | 61 +--
.../pulsar/testclient/PulsarPerfTestTool.java | 27 +-
.../pulsar/testclient/PerfClientUtilsTest.java | 23 +
.../testclient/PerformanceBaseArgumentsTest.java | 34 ++
.../coordinator/impl/MLTransactionLogImpl.java | 2 +
.../coordinator/impl/MLTransactionLogImplTest.java | 25 ++
53 files changed, 3261 insertions(+), 377 deletions(-)
create mode 100644 managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerRetentionTest.java
create mode 100644 managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabledTest.java
create mode 100644 microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerCohortBenchmark.java
create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
create mode 100644 pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ClusterUrlTest.java
(pulsar) 14/14: [fix][ml] Fix active cursor retention and deferred-update synchronization (#26511)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b3dcc415750c667d06f87fa0a0036601ff41b9dc
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 9 17:25:52 2026 +0300
[fix][ml] Fix active cursor retention and deferred-update synchronization (#26511)
(cherry picked from commit 72632e512b3714d42966fee1774613071eb89c91)
---
managed-ledger/pom.xml | 5 +
.../impl/ActiveManagedCursorContainerImpl.java | 208 +++++++--
.../ActiveManagedCursorContainerRetentionTest.java | 491 +++++++++++++++++++++
.../ActiveManagedCursorContainerBenchmark.java | 44 +-
...ctiveManagedCursorContainerCohortBenchmark.java | 82 ++++
5 files changed, 798 insertions(+), 32 deletions(-)
diff --git a/managed-ledger/pom.xml b/managed-ledger/pom.xml
index 2af2d596c29f..da1f381f1fe0 100644
--- a/managed-ledger/pom.xml
+++ b/managed-ledger/pom.xml
@@ -67,6 +67,11 @@
<artifactId>guava</artifactId>
</dependency>
+ <dependency>
+ <groupId>it.unimi.dsi</groupId>
+ <artifactId>fastutil</artifactId>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>testmocks</artifactId>
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerImpl.java
index 649f3060e9c4..889dee2e2058 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerImpl.java
@@ -20,13 +20,16 @@ package org.apache.bookkeeper.mledger.impl;
import static java.util.Objects.requireNonNull;
import com.google.common.annotations.VisibleForTesting;
+import it.unimi.dsi.fastutil.objects.ObjectArrayList;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.PriorityQueue;
+import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.StampedLock;
@@ -49,10 +52,16 @@ import org.apache.commons.lang3.tuple.Pair;
*/
@Slf4j
public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorContainer {
+ private static final int MIN_REMOVALS_BEFORE_CLEANUP = 64;
+ // Below this size, the JDK 21 object-array sort does not allocate a temporary merge buffer.
+ private static final int SMALL_POSITION_UPDATE_BATCH = 32;
+
private static class Node {
- final ManagedCursor cursor;
+ // Iterators read this without the lock. Removed nodes must not retain the cursor's object graph.
+ volatile ManagedCursor cursor;
Position position;
Position pendingPosition;
+ boolean pendingPositionUpdateQueued;
boolean pendingRemove = false;
MutableInt numberOfCursorsAtSamePositionOrBefore;
Node prev;
@@ -65,6 +74,8 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
}
// number of nodes in the double-linked list
int trackedNodeCount = 0;
+ // Conservative count: a removed node may have been reused since the last cleanup or position flush.
+ private int removalsSinceLastCleanup;
private final long continueCachingAddedEntriesAfterLastActiveCursorLeavesMillis;
private volatile long cursorRemovedTimestampMillis;
private volatile int cursorCount;
@@ -86,18 +97,20 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
// Maps a cursor to the node
private final ConcurrentMap<String, Node> cursors = new ConcurrentHashMap<>();
private final Map<String, Node> pendingRemovedCursors = new HashMap<>();
- private final PriorityQueue<Node> pendingPositionUpdates = new PriorityQueue<>(new Comparator<Node>() {
+ // Accumulate without ordering; only incremental flushes benefit from ordering by old position.
+ private final ObjectArrayList<Node> pendingPositionUpdates = new ObjectArrayList<>();
+ private static final Comparator<Node> PENDING_POSITION_COMPARATOR = new Comparator<Node>() {
@Override
public int compare(Node o1, Node o2) {
if (o1.position == null) {
- return 1; // o1 is null, should be after o2
+ return o2.position == null ? 0 : -1;
}
if (o2.position == null) {
- return -1; // o2 is null, should be after o1
+ return 1;
}
- return o2.position.compareTo(o1.position);
+ return o1.position.compareTo(o2.position);
}
- });
+ };
private final StampedLock rwLock = new StampedLock();
@@ -114,24 +127,26 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
}
node.pendingPosition = null;
} else {
- if (node.pendingPosition == null) {
- pendingPositionUpdates.add(node);
+ // Re-adding an existing cursor cancels pending untracking, just like updateCursor.
+ if (node.pendingRemove) {
+ node.pendingRemove = false;
+ pendingRemovedCursors.remove(cursor.getName());
}
- node.pendingPosition = position;
+ queuePositionUpdate(node, position);
}
} else {
if (position != null) {
node = pendingRemovedCursors.remove(cursor.getName());
if (node != null) {
node.pendingRemove = false;
+ node.cursor = cursor;
}
}
if (node == null) {
node = new Node(cursor, position);
}
if (position != null) {
- node.pendingPosition = position;
- pendingPositionUpdates.add(node);
+ queuePositionUpdate(node, position);
}
if (cursors.put(cursor.getName(), node) == null) {
cursorCount++;
@@ -259,9 +274,16 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
pendingRemovedCursors.put(name, node);
node.pendingRemove = true;
}
+ node.cursor = null;
node.pendingPosition = null;
cursorRemovedTimestampMillis = System.currentTimeMillis();
cursorCount--;
+ // Keep position updates lazy, but bound retained nodes even if nobody queries their ordering.
+ // Scaling the batch with the live set amortizes scanning it over many removals.
+ if (++removalsSinceLastCleanup >= Math.max(MIN_REMOVALS_BEFORE_CLEANUP, cursorCount)
+ || cursorCount == 0) {
+ compactRemovedNodes();
+ }
return true;
} else {
return false;
@@ -271,6 +293,46 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
}
}
+ private void compactRemovedNodes() {
+ // Preserve lazy position updates while compacting the pending list in linear time.
+ pendingPositionUpdates.removeIf(node -> {
+ if (node.cursor == null) {
+ node.pendingPositionUpdateQueued = false;
+ return true;
+ }
+ return false;
+ });
+ if (pendingRemovedCursors.values().removeIf(node -> node.cursor == null)) {
+ Node previous = null;
+ Node current = head;
+ int count = 0;
+ while (current != null) {
+ Node next = current.next;
+ if (current.cursor == null) {
+ if (previous == null) {
+ head = next;
+ } else {
+ previous.next = next;
+ }
+ if (next != null) {
+ next.prev = previous;
+ }
+ current.prev = null;
+ current.next = null;
+ trackedNodeCount--;
+ } else {
+ // Equal-position nodes already share a counter. The last survivor in each group
+ // sets its final rank, avoiding per-removal counter scans and new allocations.
+ current.numberOfCursorsAtSamePositionOrBefore.setValue(++count);
+ previous = current;
+ }
+ current = next;
+ }
+ tail = previous;
+ }
+ removalsSinceLastCleanup = 0;
+ }
+
/**
* Remove a node from the double-linked list and update counters.
*/
@@ -334,17 +396,25 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
pendingRemovedCursors.remove(cursor.getName());
}
// position changed, mark the node as pending
- if (node.pendingPosition == null) {
- pendingPositionUpdates.add(node);
- }
- node.pendingPosition = newPosition;
+ queuePositionUpdate(node, newPosition);
}
} finally {
rwLock.unlockWrite(stamp);
}
}
+ private void queuePositionUpdate(Node node, Position position) {
+ // Clearing pendingPosition cancels an update, but does not remove its pending-list entry.
+ // In particular, reactivating a node must not enqueue the same node again.
+ if (!node.pendingPositionUpdateQueued) {
+ pendingPositionUpdates.add(node);
+ node.pendingPositionUpdateQueued = true;
+ }
+ node.pendingPosition = position;
+ }
+
private void processPendingPositions() {
+ removalsSinceLastCleanup = 0;
if (pendingRemovedCursors.isEmpty() && pendingPositionUpdates.isEmpty()) {
// No pending changes, nothing to do
return;
@@ -365,11 +435,24 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
}
pendingRemovedCursors.clear();
}
- while (true) {
- Node node = pendingPositionUpdates.poll();
- if (node == null) {
- break; // No more pending updates
+ int pendingCount = pendingPositionUpdates.size();
+ if (pendingCount > 1 && pendingCount < SMALL_POSITION_UPDATE_BATCH) {
+ // Keep the efficient small-array sort without allocating a temporary merge buffer.
+ pendingPositionUpdates.sort(PENDING_POSITION_COMPARATOR);
+ } else if (pendingCount >= SMALL_POSITION_UPDATE_BATCH) {
+ // Ordered batches need no sort. Otherwise use in-place sorting to avoid per-flush buffers.
+ for (int i = 1; i < pendingCount; i++) {
+ if (PENDING_POSITION_COMPARATOR.compare(pendingPositionUpdates.get(i - 1),
+ pendingPositionUpdates.get(i)) > 0) {
+ pendingPositionUpdates.unstableSort(PENDING_POSITION_COMPARATOR);
+ break;
+ }
}
+ }
+ while (!pendingPositionUpdates.isEmpty()) {
+ // Drain from the end to process descending old positions without shifting array elements.
+ Node node = pendingPositionUpdates.remove(pendingPositionUpdates.size() - 1);
+ node.pendingPositionUpdateQueued = false;
if (node.pendingPosition != null) {
if (node.position == null) {
node.position = node.pendingPosition;
@@ -393,6 +476,7 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
// Collect all nodes that should be in the list and update their positions
List<Node> activeNodes = new ArrayList<>();
for (Node node : cursors.values()) {
+ node.pendingPositionUpdateQueued = false;
if (node.pendingPosition != null) {
node.position = node.pendingPosition;
node.pendingPosition = null;
@@ -473,12 +557,13 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
// update the position
node.position = newPosition;
- if (node == tail) {
- // Node is at the tail, no need to update counters
+ // A tail that remains in its own position group keeps the same rank and counter.
+ if (node == tail && (node.prev == null
+ || (node.prev.position.compareTo(oldPosition) != 0
+ && node.prev.position.compareTo(newPosition) != 0))) {
return;
}
-
- // update the counters
+ // Even the tail must leave or join shared counters when its position group changes.
if (movingForward) {
// first decrement the counter for the old position
node.numberOfCursorsAtSamePositionOrBefore.decrement();
@@ -684,11 +769,21 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
@Override
public Position getSlowestCursorPosition() {
- long stamp = rwLock.readLock();
+ long stamp = rwLock.tryOptimisticRead();
+ if (stamp != 0 && pendingRemovedCursors.isEmpty() && pendingPositionUpdates.isEmpty()) {
+ // Snapshot scalar fields only; a concurrent writer can invalidate the list before validation.
+ Node first = head;
+ Position position = first != null ? first.position : null;
+ if (rwLock.validate(stamp)) {
+ return position;
+ }
+ }
+ // Flushing pending changes mutates the list and must be exclusive.
+ stamp = rwLock.writeLock();
try {
return internalSlowestReaderPosition();
} finally {
- rwLock.unlockRead(stamp);
+ rwLock.unlockWrite(stamp);
}
}
@@ -725,14 +820,24 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
public Iterator<ManagedCursor> iterator() {
final Iterator<Map.Entry<String, Node>> it = cursors.entrySet().iterator();
return new Iterator<ManagedCursor>() {
+ private ManagedCursor nextCursor;
+
@Override
public boolean hasNext() {
- return it.hasNext();
+ while (nextCursor == null && it.hasNext()) {
+ nextCursor = it.next().getValue().cursor;
+ }
+ return nextCursor != null;
}
@Override
public ManagedCursor next() {
- return it.next().getValue().cursor;
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ ManagedCursor result = nextCursor;
+ nextCursor = null;
+ return result;
}
@Override
@@ -757,6 +862,33 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
}
}
+ @VisibleForTesting
+ int getPendingPositionUpdatesCount() {
+ long stamp = rwLock.readLock();
+ try {
+ return pendingPositionUpdates.size();
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
+ /** Returns one entry per retained node, including null for nodes whose cursor was released. */
+ @VisibleForTesting
+ List<ManagedCursor> getRetainedCursors() {
+ long stamp = rwLock.readLock();
+ try {
+ Set<Node> nodes = new HashSet<>(cursors.values());
+ nodes.addAll(pendingPositionUpdates);
+ nodes.addAll(pendingRemovedCursors.values());
+ for (Node node = head; node != null; node = node.next) {
+ nodes.add(node);
+ }
+ return nodes.stream().map(node -> node.cursor).toList();
+ } finally {
+ rwLock.unlockRead(stamp);
+ }
+ }
+
@Override
public int size() {
return cursorCount;
@@ -784,14 +916,21 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
*/
@VisibleForTesting
void checkOrderingAndNumberOfCursorsState() {
- long stamp = rwLock.readLock();
+ long stamp = rwLock.writeLock();
try {
processPendingPositions();
Node current = head;
+ Node previous = null;
int currentCount = 0;
Position lastPosition = null;
List<String> lastPositionCursorNames = new ArrayList<>();
while (current != null) {
+ if (current.cursor == null) {
+ throw new IllegalStateException("Released node still linked");
+ }
+ if (current.prev != previous) {
+ throw new IllegalStateException("Previous node link is inconsistent");
+ }
if (current.prev != null && current.position.compareTo(current.prev.position) < 0) {
throw new IllegalStateException("Cursors are not ordered: " + current.cursor.getName()
+ " with position " + current.position + " is after cursor " + current.prev.cursor.getName()
@@ -826,11 +965,18 @@ public class ActiveManagedCursorContainerImpl implements ActiveManagedCursorCont
} else {
lastPositionCursorNames.add(current.cursor.getName());
}
+ previous = current;
current = current.next;
-
+ }
+ if (tail != previous) {
+ throw new IllegalStateException("Tail does not match the last linked node");
+ }
+ if (currentCount != trackedNodeCount) {
+ throw new IllegalStateException("Tracked node count does not match the linked list: expected "
+ + trackedNodeCount + ", but found " + currentCount);
}
} finally {
- rwLock.unlockRead(stamp);
+ rwLock.unlockWrite(stamp);
}
}
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerRetentionTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerRetentionTest.java
new file mode 100644
index 000000000000..a89a2ea22596
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerRetentionTest.java
@@ -0,0 +1,491 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.bookkeeper.mledger.impl.MockManagedCursor.addCursor;
+import static org.apache.bookkeeper.mledger.impl.MockManagedCursor.createCursor;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ActiveManagedCursorContainerRetentionTest {
+ private static final Position POSITION = PositionFactory.create(1, 1);
+
+ @DataProvider
+ public Object[][] positionUpdateBatches() {
+ List<Object[]> cases = new ArrayList<>();
+ for (int batchSize : new int[] {1, 31, 32, 33, 80, 100}) {
+ for (boolean reverse : new boolean[] {false, true}) {
+ for (boolean includeNewCursors : new boolean[] {false, true}) {
+ cases.add(new Object[] {batchSize, reverse, includeNewCursors});
+ }
+ }
+ }
+ return cases.toArray(new Object[0][]);
+ }
+
+ @Test(dataProvider = "positionUpdateBatches")
+ public void testBatchedPositionsMatchCursorRanks(int batchSize, boolean reverse, boolean includeNewCursors) {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ Map<String, Position> positions = new HashMap<>();
+ for (int i = 0; i < 200; i++) {
+ Position position = PositionFactory.create(1, i / 3);
+ addCursor(container, "cursor" + i, position);
+ positions.put("cursor" + i, position);
+ }
+ container.getSlowestCursorPosition();
+ for (int round = 0; round < 2; round++) {
+ for (int i = 0; i < batchSize; i++) {
+ int index = reverse ? batchSize - i - 1 : i;
+ String name = includeNewCursors && index % 2 == 0 ? "new" + index : "cursor" + index;
+ Position intermediate = PositionFactory.create(1, 1000 + index);
+ if (!positions.containsKey(name)) {
+ addCursor(container, name, intermediate);
+ } else {
+ container.updateCursor(container.get(name), intermediate);
+ }
+ // Coalesce repeated updates while moving into shared groups in both directions.
+ Position latest = PositionFactory.create(1, (index + round) % 2 == 0 ? index % 7 : 200 + index % 7);
+ container.updateCursor(container.get(name), latest);
+ positions.put(name, latest);
+ }
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(batchSize);
+ for (Map.Entry<String, Position> entry : positions.entrySet()) {
+ int expectedRank = (int) positions.values().stream()
+ .filter(position -> position.compareTo(entry.getValue()) <= 0).count();
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get(entry.getKey())))
+ .as("rank of %s after batch %s", entry.getKey(), round).isEqualTo(expectedRank);
+ }
+ assertThat(container.getSlowestCursorPosition())
+ .isEqualTo(positions.values().stream().min(Position::compareTo).orElseThrow());
+ assertThat(container.getPendingPositionUpdatesCount()).isZero();
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+ }
+
+ @DataProvider
+ public Object[][] tracked() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "tracked")
+ public void testRemovedCursorReleasedBeforeCleanup(boolean tracked) {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ ManagedCursor survivor = createCursor(container, "survivor", POSITION);
+ container.add(survivor, POSITION);
+ addCursor(container, "removed", POSITION);
+ if (tracked) {
+ container.getSlowestCursorPosition();
+ }
+ container.removeCursor("removed");
+
+ // Inspect retention without triggering a position flush or relying on GC timing.
+ assertThat(container.getRetainedCursors()).containsExactlyInAnyOrder(survivor, null);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(survivor)).isEqualTo(1);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test(dataProvider = "tracked")
+ public void testLastRemovalClearsRetainedNodes(boolean tracked) {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < 100; i++) {
+ addCursor(container, "cursor", POSITION);
+ if (tracked) {
+ container.getSlowestCursorPosition();
+ }
+ assertThat(container.removeCursor("cursor")).isTrue();
+ assertThat(container.getRetainedCursors()).isEmpty();
+ assertThat(container.size()).isZero();
+ }
+ }
+
+ @Test
+ public void testPendingNodesBoundedWithoutPositionQueries() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ ManagedCursor survivor = createCursor(container, "survivor", POSITION);
+ container.add(survivor, POSITION);
+ for (int i = 0; i < 1000; i++) {
+ addCursor(container, "cursor" + i, POSITION);
+ container.removeCursor("cursor" + i);
+ List<ManagedCursor> retained = container.getRetainedCursors();
+ assertThat(retained).hasSizeLessThanOrEqualTo(65);
+ assertThat(retained.stream().filter(cursor -> cursor != null).toList()).containsExactly(survivor);
+ }
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(survivor)).isEqualTo(1);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testTrackedRemovalsTriggerCleanupWithoutPendingUpdates() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < 1000; i++) {
+ addCursor(container, "cursor" + i, PositionFactory.create(1, i));
+ }
+ container.getSlowestCursorPosition();
+ for (int i = 0; i < 900; i++) {
+ container.removeCursor("cursor" + i);
+ assertThat(container.getRetainedCursors()).hasSizeLessThanOrEqualTo(
+ container.size() + Math.max(64, container.size()));
+ if (i == 63) {
+ // A large live set should not be rebuilt every fixed-size batch of removals.
+ assertThat(container.getRetainedCursors()).hasSize(1000);
+ }
+ }
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(PositionFactory.create(1, 900));
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor999"))).isEqualTo(100);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testCompactionPreservesPendingUpdatesAndSharedCounters() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < 200; i++) {
+ addCursor(container, "cursor" + i, PositionFactory.create(1, i / 10));
+ }
+ container.getSlowestCursorPosition();
+ ManagedCursor moved = container.get("cursor199");
+ container.updateCursor(moved, PositionFactory.create(1, 0));
+ for (int i = 0; i < 200; i += 2) {
+ container.removeCursor("cursor" + i);
+ }
+ assertThat(container.getRetainedCursors()).hasSize(100).doesNotContainNull();
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(1);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(moved)).isEqualTo(6);
+ for (int i = 1; i < 199; i += 2) {
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor" + i)))
+ .as("rank of cursor%s after compaction and the pending move", i)
+ .isEqualTo(Math.min(100, (i / 10 + 1) * 5 + 1));
+ }
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testPositionQueryResetsCleanupBatch() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ addCursor(container, "survivor", POSITION);
+ for (int i = 0; i < 63; i++) {
+ addCursor(container, "cursor" + i, POSITION);
+ container.removeCursor("cursor" + i);
+ }
+ container.getSlowestCursorPosition();
+ addCursor(container, "removed", POSITION);
+ container.removeCursor("removed");
+ // The query already reclaimed the old batch; this removal should remain deferred.
+ assertThat(container.getRetainedCursors()).hasSize(2).containsNull();
+ }
+
+ @Test
+ public void testReusedNodeUsesCurrentCursorInstance() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ addCursor(container, "survivor", POSITION);
+ addCursor(container, "recreated", POSITION);
+ container.getSlowestCursorPosition();
+ container.removeCursor("recreated");
+ ManagedCursor replacement = createCursor(container, "recreated", POSITION);
+ container.add(replacement, POSITION);
+
+ assertThat(container.get("recreated")).isSameAs(replacement);
+ assertThat(container.getRetainedCursors()).doesNotContainNull().contains(replacement);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(replacement)).isEqualTo(2);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testReactivationDoesNotDuplicatePendingUpdates() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ addCursor(container, "first", POSITION);
+ addCursor(container, "second", POSITION);
+ container.getSlowestCursorPosition();
+ for (int i = 0; i < 1000; i++) {
+ container.removeCursor("first");
+ addCursor(container, "first", POSITION);
+ container.removeCursor("second");
+ addCursor(container, "second", POSITION);
+ if (i < 31) {
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(2);
+ } else {
+ // Compaction can discard the removed node before reactivation, leaving new orphan
+ // nodes until the next batch. Reused active nodes must not accumulate pending-list entries.
+ assertThat(container.getPendingPositionUpdatesCount()).isLessThanOrEqualTo(66);
+ }
+ }
+ container.checkOrderingAndNumberOfCursorsState();
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("first"))).isEqualTo(2);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("second"))).isEqualTo(2);
+ }
+
+ @DataProvider
+ public Object[][] pendingUntracking() {
+ return new Object[][] {{1, false}, {1, true}, {10, false}, {10, true}};
+ }
+
+ @Test(dataProvider = "pendingUntracking")
+ public void testAddExistingCursorCancelsPendingUntracking(int cursorCount, boolean untrackWithAdd) {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < cursorCount; i++) {
+ addCursor(container, "cursor" + i, PositionFactory.create(1, i + 1));
+ }
+ container.getSlowestCursorPosition();
+ int movedIndex = cursorCount / 2;
+ ManagedCursor moved = container.get("cursor" + movedIndex);
+ if (untrackWithAdd) {
+ container.add(moved, null);
+ } else {
+ container.updateCursor(moved, null);
+ }
+ Position later = PositionFactory.create(1, cursorCount + 1);
+ container.add(moved, later);
+ // One pending update rebuilds a singleton container but uses incremental movement with ten cursors.
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(cursorCount == 1 ? later : POSITION);
+ assertThat(container.size()).isEqualTo(cursorCount);
+ for (int i = 0; i < cursorCount; i++) {
+ int expectedRank = i == movedIndex ? cursorCount : (i < movedIndex ? i + 1 : i);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor" + i)))
+ .as("rank of cursor%s after cancelling pending untracking", i)
+ .isEqualTo(expectedRank);
+ }
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testReactivatedTailLeavesSharedPositionGroup() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < 6; i++) {
+ addCursor(container, "cursor" + i, POSITION);
+ }
+ container.getSlowestCursorPosition();
+ // Insert incrementally after the existing group to establish the tail independently of map iteration order.
+ addCursor(container, "tail", POSITION);
+ container.getSlowestCursorPosition();
+ for (int i = 3; i < 6; i++) {
+ container.removeCursor("cursor" + i);
+ }
+ container.getSlowestCursorPosition();
+
+ Position later = PositionFactory.create(1, 2);
+ container.updateCursor(container.get("tail"), later);
+ container.removeCursor("tail");
+ addCursor(container, "tail", later);
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(1);
+ for (int i = 0; i < 3; i++) {
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor" + i)))
+ .as("rank of cursor%s after the reactivated tail leaves its group", i)
+ .isEqualTo(3);
+ }
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("tail"))).isEqualTo(4);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testTailPositionGroupsRemainCorrectAfterCompaction() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 0; i < 4; i++) {
+ addCursor(container, "cursor" + i, POSITION);
+ }
+ addCursor(container, "tail", PositionFactory.create(1, 3));
+ container.getSlowestCursorPosition();
+ ManagedCursor tailCursor = container.get("tail");
+ // Join and leave the preceding group, and move in both directions within a separate tail group.
+ for (int entryId : new int[] {1, 2, 3, 2, 1, 2}) {
+ container.updateCursor(tailCursor, PositionFactory.create(1, entryId));
+ for (int i = 0; i < 4; i++) {
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor" + i)))
+ .as("rank of cursor%s with tail at entry %s", i, entryId)
+ .isEqualTo(entryId == 1 ? 5 : 4);
+ }
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(tailCursor)).isEqualTo(5);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ container.removeCursor("cursor0");
+ for (int i = 0; i < 63; i++) {
+ addCursor(container, "temporary" + i, POSITION);
+ container.removeCursor("temporary" + i);
+ }
+ // The removal batch compacts the tracked list without flushing any position updates.
+ assertThat(container.getRetainedCursors()).hasSize(4).doesNotContainNull();
+ for (int i = 1; i < 4; i++) {
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(container.get("cursor" + i)))
+ .isEqualTo(3);
+ }
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(tailCursor)).isEqualTo(4);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testCompactionRemovesTailBeforeAppending() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ for (int i = 1; i <= 5; i++) {
+ addCursor(container, "cursor" + i, PositionFactory.create(1, i));
+ }
+ container.getSlowestCursorPosition();
+ container.removeCursor("cursor5");
+ for (int i = 0; i < 63; i++) {
+ addCursor(container, "temporary" + i, POSITION);
+ container.removeCursor("temporary" + i);
+ }
+ // Verify compaction occurred without a position query repairing the list first.
+ assertThat(container.getRetainedCursors()).hasSize(4).doesNotContainNull();
+ Position appendedPosition = PositionFactory.create(1, 9);
+ addCursor(container, "appended", appendedPosition);
+ ManagedCursor appended = container.get("appended");
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(1);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(appended)).isEqualTo(5);
+ container.checkOrderingAndNumberOfCursorsState();
+ for (int i = 1; i <= 4; i++) {
+ container.removeCursor("cursor" + i);
+ }
+ assertThat(container.size()).isEqualTo(1);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(appendedPosition);
+ assertThat(container.getNumberOfCursorsAtSamePositionOrBefore(appended)).isEqualTo(1);
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Test
+ public void testSlowestPositionFlushWaitsForReaders() throws Exception {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ ManagedCursor cursor = mock(ManagedCursor.class);
+ when(cursor.getName()).thenReturn("cursor");
+ container.add(cursor, POSITION);
+ CountDownLatch readLockHeld = new CountDownLatch(1);
+ CountDownLatch releaseReadLock = new CountDownLatch(1);
+ AtomicReference<Thread> flushThread = new AtomicReference<>();
+ assertThat(container.getPendingPositionUpdatesCount()).isEqualTo(1);
+ when(cursor.toString()).thenAnswer(invocation -> {
+ readLockHeld.countDown();
+ assertThat(releaseReadLock.await(30, SECONDS)).isTrue();
+ return "cursor";
+ });
+ ExecutorService executor = Executors.newFixedThreadPool(2,
+ new DefaultThreadFactory("cursor-container-flush-test"));
+ try {
+ Future<String> reader = executor.submit(container::toString);
+ assertThat(readLockHeld.await(10, SECONDS)).isTrue();
+ Future<Position> slowest = executor.submit(() -> {
+ flushThread.set(Thread.currentThread());
+ return container.getSlowestCursorPosition();
+ });
+ // Wait for actual lock contention, rather than assuming the task ran after a fixed delay.
+ // A shared-lock regression completes the task instead and fails the following assertion.
+ await().atMost(10, SECONDS).until(() -> slowest.isDone()
+ || (flushThread.get() != null && flushThread.get().getState() == Thread.State.WAITING));
+ assertThat(slowest.isDone()).isFalse();
+ releaseReadLock.countDown();
+ assertThat(slowest.get(10, SECONDS)).isEqualTo(POSITION);
+ assertThat(reader.get(10, SECONDS)).isEqualTo("[cursor]");
+ assertThat(container.getPendingPositionUpdatesCount()).isZero();
+ } finally {
+ releaseReadLock.countDown();
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(10, SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ public void testSlowestPositionReadDoesNotExcludeOtherReaders() throws Exception {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ ManagedCursor cursor = mock(ManagedCursor.class);
+ when(cursor.getName()).thenReturn("cursor");
+ container.add(cursor, POSITION);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(POSITION);
+
+ CountDownLatch readLockHeld = new CountDownLatch(1);
+ CountDownLatch releaseReadLock = new CountDownLatch(1);
+ when(cursor.toString()).thenAnswer(invocation -> {
+ // toString calls the cursor while holding the container's read lock.
+ readLockHeld.countDown();
+ assertThat(releaseReadLock.await(30, SECONDS)).isTrue();
+ return "cursor";
+ });
+ ExecutorService executor = Executors.newFixedThreadPool(2,
+ new DefaultThreadFactory("cursor-container-read-test"));
+ try {
+ Future<String> reader = executor.submit(container::toString);
+ assertThat(readLockHeld.await(10, SECONDS)).isTrue();
+ Future<Position> slowest = executor.submit(container::getSlowestCursorPosition);
+ assertThat(slowest.get(10, SECONDS)).isEqualTo(POSITION);
+ releaseReadLock.countDown();
+ assertThat(reader.get(10, SECONDS)).isEqualTo("[cursor]");
+ } finally {
+ releaseReadLock.countDown();
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(10, SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ public void testSlowestPositionFlushesPendingChanges() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ assertThat(container.getSlowestCursorPosition()).isNull();
+ Position later = PositionFactory.create(1, 2);
+ Position latest = PositionFactory.create(1, 3);
+ addCursor(container, "later", later);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(later);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(later);
+ addCursor(container, "earlier", POSITION);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(POSITION);
+ container.updateCursor(container.get("earlier"), latest);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(later);
+ container.removeCursor("later");
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(latest);
+ assertThat(container.getSlowestCursorPosition()).isEqualTo(latest);
+ container.removeCursor("earlier");
+ assertThat(container.getSlowestCursorPosition()).isNull();
+ }
+
+ @Test
+ public void testIteratorSkipsRemovedNodes() {
+ ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl();
+ addCursor(container, "first", POSITION);
+ addCursor(container, "second", POSITION);
+ Iterator<ManagedCursor> iterator = container.iterator();
+ container.removeCursor("first");
+ container.removeCursor("second");
+ assertThat(iterator.hasNext()).isFalse();
+ assertThatThrownBy(iterator::next).isInstanceOf(NoSuchElementException.class);
+
+ addCursor(container, "third", POSITION);
+ Iterator<ManagedCursor> prefetched = container.iterator();
+ assertThat(prefetched.hasNext()).isTrue();
+ assertThat(prefetched.hasNext()).isTrue();
+ container.removeCursor("third");
+ assertThat(prefetched.next().getName()).isEqualTo("third");
+ assertThat(prefetched.hasNext()).isFalse();
+ }
+}
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerBenchmark.java
index 75aea0166fa8..a282953868d7 100644
--- a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerBenchmark.java
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerBenchmark.java
@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -132,6 +133,47 @@ public class ActiveManagedCursorContainerBenchmark {
return doRandomSeekingForward(threadState);
}
+ @Threads(1)
+ @Benchmark
+ public int tailSeekingForward(ThreadState threadState) {
+ long counter = threadState.nextCounter();
+ ManagedCursor cursor = cursors.get(cursors.size() - 1);
+ // Keep moving the fastest cursor, without joining or leaving a shared position group.
+ cursor.seek(cursor.getReadPosition().getPositionAfterEntries(1));
+ return getNumberOfCursorsAtSamePositionOrBeforeRatio > 0
+ && counter % getNumberOfCursorsAtSamePositionOrBeforeRatio == 0
+ ? container.getNumberOfCursorsAtSamePositionOrBefore(cursor)
+ : (int) cursor.getReadPosition().getEntryId();
+ }
+
+ @Threads(1)
+ @Benchmark
+ public Position slowestCursorPosition01() {
+ return container.getSlowestCursorPosition();
+ }
+
+ @Threads(10)
+ @Benchmark
+ public Position slowestCursorPosition10() {
+ return container.getSlowestCursorPosition();
+ }
+
+ @Threads(1)
+ @Benchmark
+ public int cursorChurn(ThreadState threadState) {
+ long counter = threadState.nextCounter();
+ // Keep the other cursors active while repeatedly replacing one subscription.
+ container.removeCursor(cursors.get(0).getName());
+ // Use a new name and instance so that removed nodes cannot simply be reused indefinitely.
+ MockManagedCursor cursor = MockManagedCursor.createCursor(container, "churn" + counter,
+ PositionFactory.create(0, counter));
+ cursors.set(0, cursor);
+ container.add(cursor, cursor.getReadPosition());
+ return getNumberOfCursorsAtSamePositionOrBeforeRatio > 0
+ && counter % getNumberOfCursorsAtSamePositionOrBeforeRatio == 0
+ ? container.getNumberOfCursorsAtSamePositionOrBefore(cursor) : container.size();
+ }
+
private int doRandomSeekingForward(ThreadState threadState) {
Random random = threadState.random;
long counter = threadState.nextCounter();
@@ -147,4 +189,4 @@ public class ActiveManagedCursorContainerBenchmark {
// we return the entry id of the cursor's read position
: (int) cursor.getReadPosition().getEntryId();
}
-}
\ No newline at end of file
+}
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerCohortBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerCohortBenchmark.java
new file mode 100644
index 000000000000..cf233dcb9a0c
--- /dev/null
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ActiveManagedCursorContainerCohortBenchmark.java
@@ -0,0 +1,82 @@
+/*
+ * 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.bookkeeper.mledger.impl;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Level;
+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.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+public class ActiveManagedCursorContainerCohortBenchmark {
+ @Param({"100", "500"})
+ public int numberOfCursors;
+ private ActiveManagedCursorContainerImpl container;
+ private MockManagedCursor[] cohort;
+
+ @Setup(Level.Trial)
+ public void setup() {
+ container = new ActiveManagedCursorContainerImpl();
+ cohort = new MockManagedCursor[32];
+ for (int i = 0; i < numberOfCursors; i++) {
+ var position = PositionFactory.create(1, i);
+ var cursor = MockManagedCursor.createCursor(container, "cursor" + i, position);
+ container.add(cursor, position);
+ if (i >= numberOfCursors - cohort.length) {
+ cohort[i - (numberOfCursors - cohort.length)] = cursor;
+ }
+ }
+ container.checkOrderingAndNumberOfCursorsState();
+ }
+
+ @Benchmark
+ @Threads(1)
+ @OperationsPerInvocation(32)
+ public int advanceCohort() {
+ // Queue a wave in ascending old-position order. Each cursor advances beyond all old cohort positions.
+ for (var cursor : cohort) {
+ cursor.seek(cursor.getReadPosition().getPositionAfterEntries(cohort.length));
+ }
+ // 32 pending updates stay below trackedNodeCount / 2 for both parameters, forcing incremental processing.
+ return container.getNumberOfCursorsAtSamePositionOrBefore(cohort[cohort.length - 1]);
+ }
+
+ @TearDown(Level.Trial)
+ public void verify() {
+ container.checkOrderingAndNumberOfCursorsState();
+ for (int i = 0; i < cohort.length; i++) {
+ int expected = numberOfCursors - cohort.length + i + 1;
+ if (container.getNumberOfCursorsAtSamePositionOrBefore(cohort[i]) != expected) {
+ throw new AssertionError("Wrong cohort rank at index " + i);
+ }
+ }
+ }
+}
(pulsar) 04/14: [fix][cli] Apply PULSAR_MEM and PULSAR_GC in bin/pulsar-perf (#26465)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a2ddf086e3ea7b88dfdc6355a9cf8f4712730521
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 5 01:00:53 2026 +0300
[fix][cli] Apply PULSAR_MEM and PULSAR_GC in bin/pulsar-perf (#26465)
(cherry picked from commit ffd9b6a013bb8332373cd3b40a2674310f14cd80)
---
bin/pulsar-perf | 20 ++++++++++++++++++++
1 file changed, 20 insertions(+)
diff --git a/bin/pulsar-perf b/bin/pulsar-perf
index c359ce05b316..be4eb43c38fa 100755
--- a/bin/pulsar-perf
+++ b/bin/pulsar-perf
@@ -27,6 +27,13 @@ PULSAR_LOG_DIR=${PULSAR_LOG_DIR:-"$PULSAR_HOME/logs"}
DEFAULT_CLIENT_CONF=${PULSAR_CLIENT_CONF:-"$PULSAR_HOME/conf/client.conf"}
DEFAULT_LOG_CONF=$PULSAR_HOME/conf/log4j2.yaml
+# Remember whether the caller set these, since conf/pulsar_env.sh below fills them in with the
+# broker's defaults. pulsar-perf has never applied those defaults, and a load generator that
+# suddenly reserved the broker's 2g heap could fail to start where it used to run, so only a value
+# the caller passed in is applied. See where OPTS is assembled below.
+PULSAR_MEM_FROM_CALLER=${PULSAR_MEM:+true}
+PULSAR_GC_FROM_CALLER=${PULSAR_GC:+true}
+
# Check bookkeeper env and load bkenv.sh
if [ -f "$PULSAR_HOME/conf/bkenv.sh" ]
then
@@ -157,6 +164,19 @@ OPTS="-XX:+ExitOnOutOfMemoryError -Dpulsar.allocator.exit_on_oom=true $OPTS"
OPTS="-Dio.netty.recycler.maxCapacityPerThread=4096 -Dio.netty.allocator.maxOrder=10 $OPTS"
OPTS="-cp $PULSAR_CLASSPATH $OPTS"
+# Apply PULSAR_MEM and PULSAR_GC when the caller set them. Unlike conf/pulsar_tools_env.sh, which
+# folds them into PULSAR_EXTRA_OPTS itself, conf/pulsar_env.sh only defines them, so a script that
+# sources it has to add them here or they are silently dropped — which is what used to happen.
+# Only caller-supplied values are used, so that the broker defaults conf/pulsar_env.sh fills in are
+# not imposed on a tool that has never had them; setting them by editing that file therefore does
+# not affect pulsar-perf, export them instead. PULSAR_GC_LOG is left out for the same reason:
+# pulsar_env.sh always gives it a value, so including it would make every run write GC logs.
+if [ -n "$PULSAR_MEM_FROM_CALLER" ]; then
+ OPTS="$OPTS $PULSAR_MEM"
+fi
+if [ -n "$PULSAR_GC_FROM_CALLER" ]; then
+ OPTS="$OPTS $PULSAR_GC"
+fi
OPTS="$OPTS $PULSAR_EXTRA_OPTS"
# log directory & file
(pulsar) 03/14: [fix][cli] pulsar-perf: register subcommand classes with picocli instead of instances (#26467)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 52f6296d368dfe555ee60c7d719ed5727de858cb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 5 00:57:42 2026 +0300
[fix][cli] pulsar-perf: register subcommand classes with picocli instead of instances (#26467)
(cherry picked from commit f41e620a0035e8c18a38de880c6ed993838980a6)
---
.../proxy/socket/client/PerformanceClient.java | 8 +------
.../testclient/CmdGenerateDocumentation.java | 7 +-----
.../pulsar/testclient/ManagedLedgerWriter.java | 8 +------
.../pulsar/testclient/PulsarPerfTestTool.java | 27 +++++++++++-----------
4 files changed, 16 insertions(+), 34 deletions(-)
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
index 53d588e3ba7c..34370ccf34a7 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
@@ -62,10 +62,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import picocli.CommandLine;
import picocli.CommandLine.Command;
-import picocli.CommandLine.Model.CommandSpec;
import picocli.CommandLine.Option;
import picocli.CommandLine.Parameters;
-import picocli.CommandLine.Spec;
@Command(name = "websocket-producer", description = "Test pulsar websocket producer performance.")
public class PerformanceClient extends CmdBase {
@@ -138,12 +136,8 @@ public class PerformanceClient extends CmdBase {
super("websocket-producer");
}
-
- @Spec
- CommandSpec spec;
-
public void loadArguments() {
- CommandLine commander = spec.commandLine();
+ CommandLine commander = getCommander();
if (isBlank(this.authPluginClassName) && !isBlank(this.deprecatedAuthPluginClassName)) {
this.authPluginClassName = this.deprecatedAuthPluginClassName;
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
index d2e08e2cc866..d827dce76ba8 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
@@ -27,9 +27,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.proxy.socket.client.PerformanceClient;
import picocli.CommandLine;
import picocli.CommandLine.Command;
-import picocli.CommandLine.Model.CommandSpec;
import picocli.CommandLine.Option;
-import picocli.CommandLine.Spec;
@Slf4j
@Command(name = "gen-doc", description = "Generate documentation automatically.")
@@ -42,12 +40,9 @@ public class CmdGenerateDocumentation extends CmdBase{
super("gen-doc");
}
- @Spec
- CommandSpec spec;
-
@Override
public void run() throws Exception {
- CommandLine commander = spec.commandLine();
+ CommandLine commander = getCommander();
Map<String, Class<?>> cmdClassMap = new LinkedHashMap<>();
cmdClassMap.put("produce", PerformanceProducer.class);
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
index 3d73abd803f5..a37e254a0ec6 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
@@ -62,9 +62,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import picocli.CommandLine;
import picocli.CommandLine.Command;
-import picocli.CommandLine.Model.CommandSpec;
import picocli.CommandLine.Option;
-import picocli.CommandLine.Spec;
@Command(name = "managed-ledger", description = "Write directly on managed-ledgers")
public class ManagedLedgerWriter extends CmdBase{
@@ -134,13 +132,9 @@ public class ManagedLedgerWriter extends CmdBase{
super("managed-ledger");
}
-
- @Spec
- CommandSpec spec;
-
@Override
public void run() throws Exception {
- CommandLine commander = spec.commandLine();
+ CommandLine commander = getCommander();
if (this.metadataStoreUrl == null && this.zookeeperServers == null) {
System.err.println("Metadata store address argument is required (--metadata-store)");
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java
index 826060dc6b79..3c471e1b1851 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import java.io.FileInputStream;
-import java.lang.reflect.Constructor;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
@@ -66,26 +65,26 @@ public class PulsarPerfTestTool {
prop.load(fis);
}
}
- commander.setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
-
+ // Register the command classes rather than instances: picocli builds each subcommand's spec
+ // from annotations and only instantiates the user object once that subcommand is selected.
+ // Instantiating them all here ran every command's static initializers on every invocation,
+ // which is how `pulsar-perf produce` ended up allocating the latency histograms of `consume`,
+ // `read`, `transaction` and `managed-ledger` as well.
for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
- Constructor<?> constructor = c.getValue().getDeclaredConstructor();
- constructor.setAccessible(true);
- addCommand(c.getKey(), constructor.newInstance());
+ commander.addSubcommand(c.getKey(), c.getValue());
}
+ // Both settings propagate to the subcommand hierarchy as it exists when they are called, so
+ // they must be applied after the subcommands are registered. CmdBase used to set the enum
+ // behaviour on the per-command CommandLine it built in its own constructor; that instance is
+ // no longer the one picocli parses with.
+ commander.setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ commander.setCaseInsensitiveEnumValuesAllowed(true);
+
// Remove the first argument, it's the config file path
return Arrays.copyOfRange(args, 1, args.length);
}
- private void addCommand(String name, Object o) {
- if (o instanceof CmdBase) {
- commander.addSubcommand(name, ((CmdBase) o).getCommander());
- } else {
- commander.addSubcommand(o);
- }
- }
-
public static void main(String[] args) throws Exception {
PulsarPerfTestTool tool = new PulsarPerfTestTool();
args = tool.initCommander(args);
(pulsar) 06/14: [improve][ml] Parse message metadata at entry cache insert time instead of under the wrapper write lock (#26463)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2d600cfc2c154f2960b2ac7693cc79b9cd285e8b
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Sep 8 04:10:18 2026 -0700
[improve][ml] Parse message metadata at entry cache insert time instead of under the wrapper write lock (#26463)
(cherry picked from commit ce6079887474043f4055ee6c22d2c72e456ac40f)
---
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 3 +++
.../mledger/impl/cache/RangeEntryCacheImpl.java | 6 ++++++
.../impl/cache/RangeEntryCacheImplTest.java | 25 ++++++++++++++++++++++
3 files changed, 34 insertions(+)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
index 02d148504baa..569a5135c344 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
@@ -65,6 +65,9 @@ class RangeCacheEntryWrapper {
entryWrapper.key = key;
entryWrapper.value = value;
entryWrapper.size = size;
+ // When the value already carries parsed message metadata, there's no need to lazily initialize it
+ // under the write lock on the first read
+ entryWrapper.messageMetadataInitialized = value.getMessageMetadata() != null;
// Set the timestamp to the current time in nanoseconds
// This is used for time-based eviction of entries
entryWrapper.timestampNanos = System.nanoTime();
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 352ac341e560..a68ba55eeb25 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -151,6 +151,12 @@ public class RangeEntryCacheImpl implements EntryCache {
cachedData = entry.getDataBuffer().retain();
}
+ // Parse the message metadata once at insert time so that cache reads don't have to do it lazily while
+ // holding the RangeCacheEntryWrapper write lock
+ if (entry instanceof EntryImpl entryImpl) {
+ entryImpl.initializeMessageMetadataIfNeeded(ml.getName());
+ }
+
Position position = entry.getPosition();
ReferenceCountedEntry cacheEntry =
EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler(),
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
index 42ae262e8ddb..813ecc8b6d46 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
@@ -27,7 +27,9 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -45,6 +47,8 @@ import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -94,6 +98,27 @@ public class RangeEntryCacheImplTest {
expectedReadCount = () -> 1;
}
+ @Test
+ public void testInsertParsesMessageMetadata() {
+ MessageMetadata metadata = new MessageMetadata()
+ .setProducerName("producer")
+ .setSequenceId(7)
+ .setPublishTime(123456789L);
+ ByteBuf headersAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
+ Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8));
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+ assertThat(entry.getMessageMetadata()).isNull();
+
+ assertThat(rangeEntryCache.insert(entry)).isTrue();
+
+ // the metadata is parsed once at insert time instead of lazily on the first cache read
+ assertThat(entry.getMessageMetadata()).isNotNull();
+ assertThat(entry.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ assertThat(entry.getMessageMetadata().getSequenceId()).isEqualTo(7);
+ entry.release();
+ }
+
@Test
public void testPartialCachingWithMiddleEntryInCache() {
Entry entry = EntryImpl.create(1, 50, Unpooled.EMPTY_BUFFER);
(pulsar) 11/14: [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps (#26495)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 65ca2a6efdc6c54e91f48612904ffc353653ebb3
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 9 09:26:52 2026 +0300
[fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps (#26495)
(cherry picked from commit 9e0f694a8f0d021674dc1f0ef911357ece1642dd)
---
.../bookkeeper/mledger/ManagedLedgerConfig.java | 18 ++
.../mledger/impl/cache/EntryCacheDisabled.java | 4 +-
.../bookkeeper/mledger/impl/cache/RangeCache.java | 14 +-
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 38 +--
.../mledger/impl/cache/RangeEntryCacheImpl.java | 27 +-
.../mledger/impl/cache/EntryCacheDisabledTest.java | 136 +++++++++++
.../impl/cache/RangeEntryCacheImplTest.java | 271 +++++++++++++++++++--
.../pendingack/impl/MLPendingAckStoreProvider.java | 2 +
.../pulsar/broker/transaction/TransactionTest.java | 30 +++
.../coordinator/impl/MLTransactionLogImpl.java | 2 +
.../coordinator/impl/MLTransactionLogImplTest.java | 25 ++
11 files changed, 498 insertions(+), 69 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index 8fbf1cfda0d2..7c790de98dd7 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -91,6 +91,24 @@ public class ManagedLedgerConfig {
@Getter
@Setter
private boolean cacheEvictionByExpectedReadCount = true;
+ /**
+ * Whether the entries of this managed ledger are Pulsar messages, so that an entry's payload begins with the
+ * headers that parse into a {@code MessageMetadata}.
+ *
+ * <p>This holds for the managed ledger backing a topic. It doesn't hold for managed ledgers that store some
+ * other binary format, such as the transaction log and the transaction pending ack store, whose entries can
+ * never parse into valid message metadata.
+ *
+ * <p>When this is false the entry cache doesn't parse the message metadata of the entries it caches or reads.
+ * Nothing else changes: code that needs the metadata of an individual entry can still parse it on demand.
+ *
+ * <p>Set this before the managed ledger is opened for the first time, and keep it when replacing the config
+ * of an open managed ledger: {@link ManagedLedgerFactory} caches managed ledgers by name and hands back an
+ * already open one without applying the config of the later caller.
+ */
+ @Getter
+ @Setter
+ private boolean pulsarMessageEntries = true;
@Getter
private long continueCachingAddedEntriesAfterLastActiveCursorLeavesMillis;
private int minimumBacklogCursorsForCaching = 0;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
index d7040e26fe4b..d4c6ed8f908d 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
@@ -144,7 +144,9 @@ public class EntryCacheDisabled implements EntryCache {
for (LedgerEntry e : ledgerEntries) {
// Insert the entries at the end of the list (they will be unsorted for now)
EntryImpl entry = EntryImpl.create(e, interceptor, 0);
- entry.initializeMessageMetadataIfNeeded(ml.getName());
+ if (ml.getConfig().isPulsarMessageEntries()) {
+ entry.initializeMessageMetadataIfNeeded(ml.getName());
+ }
entries.add(entry);
totalSize += entry.getLength();
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
index 858b7d3c07cb..23cad445c84c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
@@ -47,22 +47,12 @@ class RangeCache {
private final ConcurrentNavigableMap<Position, RangeCacheEntryWrapper> entries;
private final RangeCacheRemovalQueue removalQueue;
private final AtomicLong size; // Total size of values stored in cache
- private final String managedLedgerName;
/**
* Construct a new RangeCache.
*/
public RangeCache(RangeCacheRemovalQueue removalQueue) {
- this(removalQueue, null);
- }
-
- /**
- * Construct a new RangeCache.
- * @param managedLedgerName the name of the managed ledger this cache belongs to
- */
- public RangeCache(RangeCacheRemovalQueue removalQueue, String managedLedgerName) {
this.removalQueue = removalQueue;
- this.managedLedgerName = managedLedgerName;
this.entries = new ConcurrentSkipListMap<>();
this.size = new AtomicLong(0);
}
@@ -125,7 +115,7 @@ class RangeCache {
if (valueWrapper == null) {
return null;
} else {
- ReferenceCountedEntry value = valueWrapper.getValue(key, managedLedgerName);
+ ReferenceCountedEntry value = valueWrapper.getValue(key);
return getRetainedValueMatchingKey(key, value);
}
}
@@ -135,7 +125,7 @@ class RangeCache {
*/
private ReferenceCountedEntry getValueMatchingEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
ReferenceCountedEntry valueMatchingEntry =
- RangeCacheEntryWrapper.getValueMatchingMapEntry(entry, managedLedgerName);
+ RangeCacheEntryWrapper.getValueMatchingMapEntry(entry);
return getRetainedValueMatchingKey(entry.getKey(), valueMatchingEntry);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
index 569a5135c344..a6d85159c244 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
@@ -25,7 +25,6 @@ import java.util.function.Function;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ReferenceCountedEntry;
-import org.apache.bookkeeper.mledger.impl.EntryImpl;
/**
* Wrapper around the value to store in Map. This is needed to ensure that a specific instance can be removed from
@@ -48,7 +47,6 @@ class RangeCacheEntryWrapper {
long size;
long timestampNanos;
int requeueCount;
- boolean messageMetadataInitialized;
volatile boolean accessed;
private RangeCacheEntryWrapper(Recycler.Handle<RangeCacheEntryWrapper> recyclerHandle) {
@@ -65,9 +63,6 @@ class RangeCacheEntryWrapper {
entryWrapper.key = key;
entryWrapper.value = value;
entryWrapper.size = size;
- // When the value already carries parsed message metadata, there's no need to lazily initialize it
- // under the write lock on the first read
- entryWrapper.messageMetadataInitialized = value.getMessageMetadata() != null;
// Set the timestamp to the current time in nanoseconds
// This is used for time-based eviction of entries
entryWrapper.timestampNanos = System.nanoTime();
@@ -80,13 +75,12 @@ class RangeCacheEntryWrapper {
/**
* Get the value associated with the key. Returns null if the key does not match the key.
*
- * @param key the key to match
- * @param managedLedgerName
+ * @param key the key to match
* @return the value associated with the key, or null if the value has already been recycled or the key does not
* match
*/
- ReferenceCountedEntry getValue(Position key, String managedLedgerName) {
- return getValueInternal(key, false, managedLedgerName);
+ ReferenceCountedEntry getValue(Position key) {
+ return getValueInternal(key, false);
}
/**
@@ -96,9 +90,8 @@ class RangeCacheEntryWrapper {
* @return the value associated with the key, or null if the value has already been recycled or the key does not
* exactly match the same instance
*/
- static ReferenceCountedEntry getValueMatchingMapEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry,
- String managedLedgerName) {
- return entry.getValue().getValueInternal(entry.getKey(), true, managedLedgerName);
+ static ReferenceCountedEntry getValueMatchingMapEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
+ return entry.getValue().getValueInternal(entry.getKey(), true);
}
/**
@@ -110,20 +103,16 @@ class RangeCacheEntryWrapper {
* key as the one stored in the wrapper. This is used to avoid any races
* when retrieving or removing the entries from the cache when the key and value
* instances are available.
- * @param managedLedgerName
* @return the value associated with the key, or null if the key does not match
*/
- private ReferenceCountedEntry getValueInternal(Position key, boolean requireSameKeyInstance,
- String managedLedgerName) {
+ private ReferenceCountedEntry getValueInternal(Position key, boolean requireSameKeyInstance) {
long stamp = lock.tryOptimisticRead();
Position localKey = this.key;
ReferenceCountedEntry localValue = this.value;
- boolean messageMetadataInitialized = this.messageMetadataInitialized;
if (!lock.validate(stamp)) {
stamp = lock.readLock();
localKey = this.key;
localValue = this.value;
- messageMetadataInitialized = this.messageMetadataInitialized;
lock.unlockRead(stamp);
}
// check that the given key matches the key associated with the value in the entry
@@ -133,20 +122,6 @@ class RangeCacheEntryWrapper {
if (localKey != key && (requireSameKeyInstance || localKey == null || !localKey.equals(key))) {
return null;
}
- // Initialize the metadata if it's not already initialized
- if (localValue != null && !messageMetadataInitialized) {
- localValue = withWriteLock(wrapper -> {
- // ensure that the key still matches
- if (wrapper.key != key && (requireSameKeyInstance || wrapper.key == null || !wrapper.key.equals(key))) {
- return null;
- }
- if (wrapper.value instanceof EntryImpl entry && !this.messageMetadataInitialized) {
- entry.initializeMessageMetadataIfNeeded(managedLedgerName);
- this.messageMetadataInitialized = true;
- }
- return wrapper.value;
- });
- }
accessed = true;
return localValue;
}
@@ -194,7 +169,6 @@ class RangeCacheEntryWrapper {
size = 0;
timestampNanos = 0;
requeueCount = 0;
- messageMetadataInitialized = false;
accessed = false;
recyclerHandle.recycle(this);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index a68ba55eeb25..56fa9a93876d 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -121,6 +121,11 @@ public class RangeEntryCacheImpl implements EntryCache {
return ml.getConfig();
}
+ @VisibleForTesting
+ RangeCache getEntries() {
+ return entries;
+ }
+
@Override
public String getName() {
return ml.getName();
@@ -151,16 +156,18 @@ public class RangeEntryCacheImpl implements EntryCache {
cachedData = entry.getDataBuffer().retain();
}
- // Parse the message metadata once at insert time so that cache reads don't have to do it lazily while
- // holding the RangeCacheEntryWrapper write lock
- if (entry instanceof EntryImpl entryImpl) {
- entryImpl.initializeMessageMetadataIfNeeded(ml.getName());
- }
-
Position position = entry.getPosition();
- ReferenceCountedEntry cacheEntry =
+ // A MessageMetadata instance keeps a reference to the buffer it was parsed from and decodes its string and
+ // bytes fields from it lazily, so it may only be shared with the cached entry when that entry keeps the
+ // same buffer alive. When the payload is copied into a cache owned buffer, the source buffer is released
+ // while the cached entry is still in the cache, so the metadata has to be parsed from the copy instead.
+ EntryImpl cacheEntry =
EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler(),
- entry.getMessageMetadata());
+ copyEntries ? null : entry.getMessageMetadata());
+ if (ml.getConfig().isPulsarMessageEntries()) {
+ // Parse the message metadata once at insert time so that cache reads don't have to do it lazily
+ cacheEntry.initializeMessageMetadataIfNeeded(ml.getName());
+ }
cachedData.release();
if (entries.put(position, cacheEntry, entryLength)) {
totalAddedEntriesSize.add(entryLength);
@@ -543,7 +550,9 @@ public class RangeEntryCacheImpl implements EntryCache {
final List<Entry> entriesToReturn = new ArrayList<>(entriesToRead);
for (LedgerEntry e : ledgerEntries) {
EntryImpl entry = EntryImpl.create(e, interceptor, expectedReadCountVal);
- entry.initializeMessageMetadataIfNeeded(ml.getName());
+ if (ml.getConfig().isPulsarMessageEntries()) {
+ entry.initializeMessageMetadataIfNeeded(ml.getName());
+ }
entriesToReturn.add(entry);
totalSize += entry.getLength();
if (expectedReadCountVal > 0) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabledTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabledTest.java
new file mode 100644
index 000000000000..65e03b46fc6b
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabledTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.bookkeeper.mledger.impl.cache;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class EntryCacheDisabledTest {
+ private ManagedLedgerImpl mockManagedLedger;
+ private ManagedLedgerConfig managedLedgerConfig;
+ private ReadHandle lh;
+
+ @BeforeMethod
+ public void setup() {
+ mockManagedLedger = mock(ManagedLedgerImpl.class);
+ when(mockManagedLedger.getName()).thenReturn("testManagedLedger");
+ managedLedgerConfig = new ManagedLedgerConfig();
+ when(mockManagedLedger.getConfig()).thenReturn(managedLedgerConfig);
+ when(mockManagedLedger.getMbean()).thenReturn(mock(ManagedLedgerMBeanImpl.class));
+ // a same-thread executor, so the read completes before the assertions run
+ when(mockManagedLedger.getExecutor()).thenReturn(MoreExecutors.newDirectExecutorService());
+ when(mockManagedLedger.getOptionalLedgerInfo(1L)).thenReturn(Optional.empty());
+ ManagedLedgerFactoryImpl mockFactory = mock(ManagedLedgerFactoryImpl.class);
+ when(mockFactory.getMbean()).thenReturn(mock(ManagedLedgerFactoryMBeanImpl.class));
+ when(mockManagedLedger.getFactory()).thenReturn(mockFactory);
+ lh = mock(ReadHandle.class);
+ when(lh.getId()).thenReturn(1L);
+ }
+
+ private static ByteBuf serializeMessage(String producerName) {
+ MessageMetadata metadata = new MessageMetadata()
+ .setProducerName(producerName)
+ .setSequenceId(7)
+ .setPublishTime(123456789L);
+ ByteBuf payload = Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8);
+ try {
+ // serializeMetadataAndPayload copies the payload instead of taking ownership of it
+ return Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata, payload);
+ } finally {
+ payload.release();
+ }
+ }
+
+ @Test
+ public void testReadDoesNotParseMessageMetadataWhenTheEntriesArentPulsarMessages() throws Exception {
+ EntryCacheDisabled entryCache = new EntryCacheDisabled(mockManagedLedger);
+
+ // the transaction log and the pending ack store keep entries that are not Pulsar messages, so the read
+ // must not try to parse message metadata out of them
+ managedLedgerConfig.setPulsarMessageEntries(false);
+ Entry entryWithoutParsing = readSingleEntry(entryCache, 0L);
+ assertThat(entryWithoutParsing.getMessageMetadata()).isNull();
+ entryWithoutParsing.release();
+
+ // control: the same bytes read over the same path do get parsed when the entries are Pulsar messages, so
+ // the assertion above can't pass merely because the payload happens to be unparseable
+ managedLedgerConfig.setPulsarMessageEntries(true);
+ Entry entryWithParsing = readSingleEntry(entryCache, 1L);
+ assertThat(entryWithParsing.getMessageMetadata()).isNotNull();
+ assertThat(entryWithParsing.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ entryWithParsing.release();
+ }
+
+ /**
+ * Reads a single freshly serialized Pulsar message back through the cache.
+ *
+ * @apiNote the returned entry must be released by the caller
+ */
+ private Entry readSingleEntry(EntryCacheDisabled entryCache, long entryId) throws Exception {
+ ByteBuf headersAndPayload = serializeMessage("producer");
+ LedgerEntryImpl ledgerEntry =
+ LedgerEntryImpl.create(1L, entryId, headersAndPayload.readableBytes(), headersAndPayload);
+ LedgerEntries ledgerEntries = mock(LedgerEntries.class);
+ when(ledgerEntries.iterator()).thenReturn(List.<LedgerEntry>of(ledgerEntry).iterator());
+ when(lh.readAsync(entryId, entryId)).thenReturn(CompletableFuture.completedFuture(ledgerEntries));
+
+ CompletableFuture<Entry> future = new CompletableFuture<>();
+ entryCache.asyncReadEntry(lh, PositionFactory.create(1L, entryId), new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ future.complete(entry);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+ Entry entry = future.get(10, TimeUnit.SECONDS);
+ // the LedgerEntries mock doesn't release what it holds, so that reference is dropped explicitly instead
+ ledgerEntry.close();
+ return entry;
+ }
+}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
index 813ecc8b6d46..5ca6c7e21fd1 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
@@ -24,6 +24,7 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -34,6 +35,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import org.apache.bookkeeper.client.api.LedgerEntries;
@@ -42,7 +44,10 @@ import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.ReferenceCountedEntry;
import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
@@ -54,20 +59,26 @@ import org.testng.annotations.Test;
public class RangeEntryCacheImplTest {
private RangeEntryCacheImpl rangeEntryCache;
+ private RangeEntryCacheManagerImpl mockEntryCacheManager;
+ private ManagedLedgerImpl mockManagedLedger;
+ private RangeCacheRemovalQueue mockRangeCacheRemovalQueue;
+ private ManagedLedgerConfig managedLedgerConfig;
private PendingReadsManager pendingReadsManager;
private ReadHandle lh;
private IntSupplier expectedReadCount;
@BeforeMethod
public void setup() {
- RangeEntryCacheManagerImpl mockEntryCacheManager = mock(RangeEntryCacheManagerImpl.class);
+ mockEntryCacheManager = mock(RangeEntryCacheManagerImpl.class);
ManagedLedgerFactoryMBeanImpl mlFactoryMBean = mock(ManagedLedgerFactoryMBeanImpl.class);
when(mockEntryCacheManager.getMlFactoryMBean()).thenReturn(mlFactoryMBean);
- ManagedLedgerImpl mockManagedLedger = mock(ManagedLedgerImpl.class);
+ mockManagedLedger = mock(ManagedLedgerImpl.class);
ManagedLedgerMBeanImpl mockManagedLedgerMBean = mock(ManagedLedgerMBeanImpl.class);
when(mockManagedLedger.getMbean()).thenReturn(mockManagedLedgerMBean);
when(mockManagedLedger.getName()).thenReturn("testManagedLedger");
- RangeCacheRemovalQueue mockRangeCacheRemovalQueue = mock(RangeCacheRemovalQueue.class);
+ managedLedgerConfig = new ManagedLedgerConfig();
+ when(mockManagedLedger.getConfig()).thenReturn(managedLedgerConfig);
+ mockRangeCacheRemovalQueue = mock(RangeCacheRemovalQueue.class);
when(mockRangeCacheRemovalQueue.addEntry(any())).thenReturn(true);
InflightReadsLimiter inflightReadsLimiter = mock(InflightReadsLimiter.class);
when(mockEntryCacheManager.getInflightReadsLimiter()).thenReturn(inflightReadsLimiter);
@@ -90,33 +101,262 @@ public class RangeEntryCacheImplTest {
callback.readEntriesComplete(entries, ctx);
return null;
}).when(pendingReadsManager).readEntries(any(), anyLong(), anyLong(), any(), any(), any());
- rangeEntryCache =
- new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, false, mockRangeCacheRemovalQueue,
- EntryLengthFunction.DEFAULT, pendingReadsManager);
+ rangeEntryCache = createRangeEntryCache(false);
lh = mock(ReadHandle.class);
when(lh.getId()).thenReturn(1L);
expectedReadCount = () -> 1;
}
- @Test
- public void testInsertParsesMessageMetadata() {
+ private RangeEntryCacheImpl createRangeEntryCache(boolean copyEntries) {
+ return new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, copyEntries,
+ mockRangeCacheRemovalQueue, EntryLengthFunction.DEFAULT, pendingReadsManager);
+ }
+
+ private static ByteBuf serializeMessage(String producerName) {
MessageMetadata metadata = new MessageMetadata()
- .setProducerName("producer")
+ .setProducerName(producerName)
.setSequenceId(7)
.setPublishTime(123456789L);
- ByteBuf headersAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
- Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8));
+ ByteBuf payload = Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8);
+ try {
+ // serializeMetadataAndPayload copies the payload instead of taking ownership of it
+ return Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata, payload);
+ } finally {
+ payload.release();
+ }
+ }
+
+ @Test
+ public void testInsertParsesMessageMetadata() {
+ ByteBuf headersAndPayload = serializeMessage("producer");
EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
headersAndPayload.release();
assertThat(entry.getMessageMetadata()).isNull();
assertThat(rangeEntryCache.insert(entry)).isTrue();
+ entry.release();
+
+ // the metadata is parsed once at insert time. Reading the entry back out of the cache doesn't parse
+ // anything any more, so this asserts what insert actually stored
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ assertThat(cached.getMessageMetadata()).isNotNull();
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
+ cached.release();
+ }
+
+ @Test
+ public void testInsertReusesTheMessageMetadataTheEntryAlreadyCarries() {
+ ByteBuf headersAndPayload = serializeMessage("in-buffer");
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+ // deliberately disagrees with the buffer, so that a silent re-parse cannot pass this test
+ MessageMetadata suppliedByTheCaller = new MessageMetadata()
+ .setProducerName("from-publish-path")
+ .setSequenceId(7)
+ .setPublishTime(123456789L);
+ entry.setMessageMetadata(suppliedByTheCaller);
+
+ assertThat(rangeEntryCache.insert(entry)).isTrue();
+ entry.release();
+
+ // the cached entry reuses the instance instead of parsing the same bytes a second time
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ assertThat(cached.getMessageMetadata()).isSameAs(suppliedByTheCaller);
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("from-publish-path");
+ cached.release();
+ }
+
+ @Test
+ public void testCachedEntryMetadataStaysReadableWhenEntriesAreCopied() {
+ RangeEntryCacheImpl copyingCache = createRangeEntryCache(true);
+ ByteBuf headersAndPayload = serializeMessage("producer");
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+
+ assertThat(copyingCache.insert(entry)).isTrue();
+ // the source entry, and with it the buffer the metadata was parsed from, is released right after the
+ // insert, exactly as OpAddEntry does on the write path
+ entry.release();
+ assertThat(headersAndPayload.refCnt()).isZero();
+
+ ReferenceCountedEntry cached = copyingCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ // MessageMetadata decodes its string and bytes fields lazily from the buffer it was parsed from, so the
+ // cached entry must not share metadata that was parsed from the now released source buffer
+ assertThat(cached.getMessageMetadata()).isNotNull();
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
+ cached.release();
+ }
+
+ @Test
+ public void testReadFromStorageDoesNotShareSourceMetadataWithTheCopiedCacheEntry() {
+ RangeEntryCacheImpl copyingCache = createRangeEntryCache(true);
+ when(mockManagedLedger.getExecutor()).thenReturn(mock(ExecutorService.class));
+ // without ledger info, ReadEntryUtils reads through ReadHandle#readAsync
+ when(mockManagedLedger.getOptionalLedgerInfo(1L)).thenReturn(Optional.empty());
+
+ ByteBuf headersAndPayload = serializeMessage("producer");
+ LedgerEntryImpl ledgerEntry =
+ LedgerEntryImpl.create(1L, 0L, headersAndPayload.readableBytes(), headersAndPayload);
+ LedgerEntries ledgerEntries = mock(LedgerEntries.class);
+ when(ledgerEntries.iterator()).thenReturn(List.<LedgerEntry>of(ledgerEntry).iterator());
+ when(lh.readAsync(0L, 0L)).thenReturn(CompletableFuture.completedFuture(ledgerEntries));
+
+ CompletableFuture<List<Entry>> future = copyingCache.readFromStorage(lh, 0L, 0L, expectedReadCount);
+ assertThat(future).isCompleted();
+ List<Entry> readEntries = future.getNow(null);
+ assertThat(readEntries).hasSize(1);
+ Entry sourceEntry = readEntries.get(0);
+
+ // unlike the write path, the read path parses the metadata of the entry it returns before inserting it,
+ // so this is the case where insert receives an entry that already carries a MessageMetadata. Only the
+ // eagerly decoded sequenceId is read from it here, since reading a string field would decode it from
+ // the buffer and keep the decoded value, hiding the very problem this test is about
+ MessageMetadata sourceMetadata = sourceEntry.getMessageMetadata();
+ assertThat(sourceMetadata).isNotNull();
+ assertThat(sourceMetadata.getSequenceId()).isEqualTo(7);
+
+ ReferenceCountedEntry cached = copyingCache.getEntries().get(PositionFactory.create(1, 0));
+ assertThat(cached).isNotNull();
+ // the cached entry is backed by a copy of the payload, so it must not share the metadata that was parsed
+ // from the source buffer
+ assertThat(cached.getMessageMetadata()).isNotNull().isNotSameAs(sourceMetadata);
+
+ // overwrite the source payload while it is still referenced, the way the pooled buffer behind it gets
+ // overwritten once it has been recycled. This turns a leftover dependency on the source buffer into a
+ // wrong value rather than into a read that only fails when the released memory happens to be reused
+ headersAndPayload.setZero(headersAndPayload.readerIndex(), headersAndPayload.readableBytes());
+ // metadata parsed from the source buffer does decode the overwritten bytes, which keeps the assertions
+ // below from turning vacuous should MessageMetadata ever stop decoding these fields lazily
+ assertThat(sourceMetadata.getProducerName()).isNotEqualTo("producer");
+
+ // the read path releases the entries it returned once dispatch is done, while the cached copy stays
+ sourceEntry.release();
+ // readFromStorage closes the LedgerEntries, but that is a mock here, so the reference the read result
+ // holds is dropped explicitly instead
+ ledgerEntry.close();
+ assertThat(headersAndPayload.refCnt()).isZero();
- // the metadata is parsed once at insert time instead of lazily on the first cache read
- assertThat(entry.getMessageMetadata()).isNotNull();
- assertThat(entry.getMessageMetadata().getProducerName()).isEqualTo("producer");
- assertThat(entry.getMessageMetadata().getSequenceId()).isEqualTo(7);
+ // MessageMetadata decodes its string and bytes fields lazily from the buffer it was parsed from, so the
+ // cached entry stays readable only because its metadata was parsed from the buffer the cache owns
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
+ cached.release();
+ }
+
+ @Test
+ public void testInsertDoesNotParseMessageMetadataWhenTheEntriesArentPulsarMessages() {
+ // the transaction log and the pending ack store keep entries that are not Pulsar messages, so the entry
+ // cache must not try to parse message metadata out of them
+ managedLedgerConfig.setPulsarMessageEntries(false);
+
+ ByteBuf headersAndPayload = serializeMessage("producer");
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+
+ assertThat(rangeEntryCache.insert(entry)).isTrue();
entry.release();
+
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ assertThat(cached.getMessageMetadata()).isNull();
+ cached.release();
+
+ // reading the entry back through the cache must not parse it either. This is what pins the removal of
+ // the lazy initialization that RangeCacheEntryWrapper used to do under its write lock, which would have
+ // defeated skipping the parse at insert time
+ Entry readBack = readSingleEntryFromCache(1, 50);
+ assertThat(readBack.getMessageMetadata()).isNull();
+ readBack.release();
+ // the read has to have been served from the cache. A miss would fall through to the mocked storage,
+ // which hands back an entry that carries no metadata either, making the assertion above vacuous
+ verify(pendingReadsManager, never()).readEntries(any(), anyLong(), anyLong(), any(), any(), any());
+
+ // control: the very same bytes are parsed when the managed ledger does hold Pulsar messages, so the
+ // assertions above can't pass merely because the payload happens to be unparseable
+ managedLedgerConfig.setPulsarMessageEntries(true);
+ ByteBuf controlHeadersAndPayload = serializeMessage("producer");
+ EntryImpl controlEntry = EntryImpl.create(1, 51, controlHeadersAndPayload);
+ controlHeadersAndPayload.release();
+ assertThat(rangeEntryCache.insert(controlEntry)).isTrue();
+ controlEntry.release();
+
+ ReferenceCountedEntry cachedControl = rangeEntryCache.getEntries().get(PositionFactory.create(1, 51));
+ assertThat(cachedControl).isNotNull();
+ assertThat(cachedControl.getMessageMetadata()).isNotNull();
+ assertThat(cachedControl.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ cachedControl.release();
+ }
+
+ @Test
+ public void testReadFromStorageDoesNotParseMessageMetadataWhenTheEntriesArentPulsarMessages() {
+ when(mockManagedLedger.getExecutor()).thenReturn(mock(ExecutorService.class));
+ // without ledger info, ReadEntryUtils reads through ReadHandle#readAsync
+ when(mockManagedLedger.getOptionalLedgerInfo(1L)).thenReturn(Optional.empty());
+ managedLedgerConfig.setPulsarMessageEntries(false);
+
+ Entry entryWithoutParsing = readSingleEntryFromStorage(0L);
+ assertThat(entryWithoutParsing.getMessageMetadata()).isNull();
+ entryWithoutParsing.release();
+
+ // control: the same bytes read over the same path do get parsed when the entries are Pulsar messages
+ managedLedgerConfig.setPulsarMessageEntries(true);
+ Entry entryWithParsing = readSingleEntryFromStorage(1L);
+ assertThat(entryWithParsing.getMessageMetadata()).isNotNull();
+ assertThat(entryWithParsing.getMessageMetadata().getProducerName()).isEqualTo("producer");
+ entryWithParsing.release();
+ }
+
+ /**
+ * Reads a single entry back through the cache read path, which is the caller that used to trigger the lazy
+ * metadata initialization inside {@link RangeCacheEntryWrapper}.
+ *
+ * @apiNote the returned entry must be released by the caller
+ */
+ private Entry readSingleEntryFromCache(long ledgerId, long entryId) {
+ CompletableFuture<Entry> future = new CompletableFuture<>();
+ rangeEntryCache.asyncReadEntry(lh, PositionFactory.create(ledgerId, entryId),
+ new AsyncCallbacks.ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ future.complete(entry);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+ assertThat(future).isCompleted();
+ return future.getNow(null);
+ }
+
+ /**
+ * Reads a single freshly serialized Pulsar message through {@link RangeEntryCacheImpl#readFromStorage}.
+ *
+ * @apiNote the returned entry must be released by the caller
+ */
+ private Entry readSingleEntryFromStorage(long entryId) {
+ ByteBuf headersAndPayload = serializeMessage("producer");
+ LedgerEntryImpl ledgerEntry =
+ LedgerEntryImpl.create(1L, entryId, headersAndPayload.readableBytes(), headersAndPayload);
+ LedgerEntries ledgerEntries = mock(LedgerEntries.class);
+ when(ledgerEntries.iterator()).thenReturn(List.<LedgerEntry>of(ledgerEntry).iterator());
+ when(lh.readAsync(entryId, entryId)).thenReturn(CompletableFuture.completedFuture(ledgerEntries));
+
+ CompletableFuture<List<Entry>> future = rangeEntryCache.readFromStorage(lh, entryId, entryId,
+ expectedReadCount);
+ assertThat(future).isCompleted();
+ List<Entry> readEntries = future.getNow(null);
+ assertThat(readEntries).hasSize(1);
+ // readFromStorage closes the LedgerEntries, but that is a mock here, so the reference the read result
+ // holds is dropped explicitly instead
+ ledgerEntry.close();
+ return readEntries.get(0);
}
@Test
@@ -221,6 +461,7 @@ public class RangeEntryCacheImplTest {
ManagedLedgerMBeanImpl mockManagedLedgerMBean = mock(ManagedLedgerMBeanImpl.class);
when(mockManagedLedger.getMbean()).thenReturn(mockManagedLedgerMBean);
when(mockManagedLedger.getName()).thenReturn("testManagedLedger");
+ when(mockManagedLedger.getConfig()).thenReturn(new ManagedLedgerConfig());
when(mockManagedLedger.getExecutor()).thenReturn(mock(java.util.concurrent.ExecutorService.class));
when(mockManagedLedger.getOptionalLedgerInfo(1L)).thenReturn(Optional.empty());
RangeCacheRemovalQueue mockRangeCacheRemovalQueue = mock(RangeCacheRemovalQueue.class);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
index 12f761bb4df5..8c55fe655422 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
@@ -155,6 +155,8 @@ public class MLPendingAckStoreProvider implements TransactionPendingAckStoreProv
Timer brokerClientSharedTimer,
PersistentTopic originPersistentTopic) {
config.setCreateIfMissing(true);
+ // the pending ack store keeps PendingAckMetadataEntry records, not Pulsar messages
+ config.setPulsarMessageEntries(false);
brokerService
.getManagedLedgerFactoryForTopic(topicName, config.getStorageClassName())
.asyncOpen(pendingAckTopicNameObject.getPersistenceNamingEncoding(),
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
index c3f969bf5f77..e2317d4a8279 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -72,6 +72,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.util.Bytes;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
@@ -604,6 +605,35 @@ public class TransactionTest extends TransactionTestBase {
});
}
+ @Test
+ public void testPendingAckStoreEntriesArentPulsarMessages() throws Exception {
+ String topicName = TopicName.get(NAMESPACE1 + "/" + "testPendingAckStoreEntriesArentPulsarMessages")
+ .toString();
+ String subName = "test";
+ // acknowledging inside a transaction initializes the pending ack store, and with it its managed ledger
+ @Cleanup
+ Consumer<byte[]> consumer = getConsumer(topicName, subName);
+ Transaction transaction = pulsarClient.newTransaction()
+ .withTransactionTimeout(10, TimeUnit.SECONDS).build().get();
+ try {
+ consumer.acknowledgeAsync(new MessageIdImpl(10, 10, 10), transaction).get();
+ } catch (ExecutionException e) {
+ // the acknowledged message id doesn't exist, which is enough to initialize the store
+ assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException);
+ }
+
+ PersistentTopic originPersistentTopic = (PersistentTopic) getPulsarServiceList().get(0)
+ .getBrokerService().getTopic(topicName, false).get().get();
+ PersistentSubscription subscription = originPersistentTopic.getSubscription(subName);
+
+ // the pending ack store keeps PendingAckMetadataEntry records, which can never parse as message metadata,
+ // so its managed ledger must be marked as not holding Pulsar messages
+ ManagedLedger pendingAckManagedLedger = subscription.getPendingAckManageLedger().get();
+ assertFalse(pendingAckManagedLedger.getConfig().isPulsarMessageEntries());
+ // control: the topic's own managed ledger does hold Pulsar messages, which is the default
+ assertTrue(originPersistentTopic.getManagedLedger().getConfig().isPulsarMessageEntries());
+ }
+
@Test
public void testTakeSnapshotBeforeBuildTxnProducer() throws Exception {
String topic = "persistent://" + NAMESPACE1 + "/testSnapShot";
diff --git a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
index 5de9b7f8fc7b..967664256301 100644
--- a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
+++ b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
@@ -99,6 +99,8 @@ public class MLTransactionLogImpl implements TransactionLog {
if (txnLogBufferedWriterConfig.isBatchEnabled()) {
this.managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled(true);
}
+ // the transaction log keeps TransactionMetadataEntry records, not Pulsar messages
+ this.managedLedgerConfig.setPulsarMessageEntries(false);
this.entryQueue = new SpscArrayQueue<>(2000);
this.bufferedWriterMetrics = bufferedWriterMetrics;
}
diff --git a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImplTest.java b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImplTest.java
index 052b2d193b9f..34660cffe8e6 100644
--- a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImplTest.java
+++ b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImplTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.transaction.coordinator.impl;
import static org.apache.pulsar.transaction.coordinator.impl.DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import com.google.common.collect.ComparisonChain;
import io.netty.util.HashedWheelTimer;
@@ -65,6 +66,30 @@ public class MLTransactionLogImplTest extends MockedBookKeeperTestCase {
{false, true}
};
}
+ @Test
+ public void testTransactionLogEntriesArentPulsarMessages() throws Exception {
+ // control: a plain managed ledger holds Pulsar messages, which is the default and what pins that the
+ // assertion below is about the transaction log rather than about the default
+ assertThat(new ManagedLedgerConfig().isPulsarMessageEntries()).isTrue();
+
+ HashedWheelTimer transactionTimer = new HashedWheelTimer(new DefaultThreadFactory("transaction-timer"),
+ 1, TimeUnit.MILLISECONDS);
+ ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
+ MLTransactionLogImpl transactionLog = new MLTransactionLogImpl(TransactionCoordinatorID.get(0), factory,
+ managedLedgerConfig, new TxnLogBufferedWriterConfig(), transactionTimer,
+ DISABLED_BUFFERED_WRITER_METRICS);
+ try {
+ transactionLog.initialize().get(3, TimeUnit.SECONDS);
+ // the transaction log stores TransactionMetadataEntry records, which can never parse as message
+ // metadata, so the entry cache must not try
+ assertThat(managedLedgerConfig.isPulsarMessageEntries()).isFalse();
+ assertThat(transactionLog.getManagedLedger().getConfig().isPulsarMessageEntries()).isFalse();
+ } finally {
+ transactionLog.closeAsync().get(3, TimeUnit.SECONDS);
+ transactionTimer.stop();
+ }
+ }
+
/**
* 1. Add some transaction logs.
* 2. Create a new transaction meta store and execute recover, assert that the txn-mapping built by Recover is as
(pulsar) 13/14: [fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4d59e8d33cc9f4c5f9d0b6b7820311dd81964574
Author: Kai Wang <kw...@apache.org>
AuthorDate: Wed Sep 9 15:34:26 2026 +0800
[fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
(cherry picked from commit d2bc2076a28a191c930f61a14fa88621b4e98400)
---
.../impl/AbstractSnapshotAbortedTxnProcessor.java | 186 +++++++++++++
.../SingleSnapshotAbortedTxnProcessorImpl.java | 44 +--
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 96 ++++---
.../buffer/impl/TopicTransactionBuffer.java | 212 +++++++++++----
.../pulsar/broker/transaction/TransactionTest.java | 6 +-
.../AbstractSnapshotAbortedTxnProcessorTest.java | 302 +++++++++++++++++++++
...napshotSegmentAbortedTxnProcessorCloseTest.java | 173 ++++++++++++
.../impl/TopicTransactionBufferCloseTest.java | 276 +++++++++++++++++++
.../impl/TopicTransactionBufferRecoveryTest.java | 88 ++++++
9 files changed, 1257 insertions(+), 126 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java
new file mode 100644
index 000000000000..2f315b977b3e
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java
@@ -0,0 +1,186 @@
+/*
+ * 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.transaction.buffer.impl;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Coordinates snapshot recovery and resource closure so resources remain available until processor-owned recovery
+ * work finishes.
+ */
+abstract class AbstractSnapshotAbortedTxnProcessor implements AbortedTxnProcessor {
+
+ private enum State {
+ IDLE,
+ RECOVERY_QUEUED,
+ RECOVERY_RUNNING,
+ CLOSED
+ }
+
+ private final ExecutorService recoveryExecutor;
+
+ private volatile State state = State.IDLE;
+ private Future<?> recoveryTask;
+ private CompletableFuture<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ // Completes before the recovery result, so close waits only for processor-owned work.
+ private CompletableFuture<Void> recoveryWorkFinishedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ AbstractSnapshotAbortedTxnProcessor(ExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public final CompletableFuture<Position> recoverFromSnapshot() {
+ CompletableFuture<Position> newRecoveryFuture;
+ CompletableFuture<Void> newRecoveryWorkFinishedFuture;
+ RejectedExecutionException submissionFailure = null;
+ synchronized (this) {
+ if (this.state == State.CLOSED) {
+ return CompletableFuture.failedFuture(closedException());
+ }
+ if (!recoveryFuture.isDone()) {
+ return recoveryFuture.copy();
+ }
+ // Bind the task to this recovery attempt so a later retry cannot complete the wrong future.
+ newRecoveryFuture = new CompletableFuture<>();
+ newRecoveryWorkFinishedFuture = new CompletableFuture<>();
+ this.recoveryFuture = newRecoveryFuture;
+ this.recoveryWorkFinishedFuture = newRecoveryWorkFinishedFuture;
+ this.state = State.RECOVERY_QUEUED;
+ try {
+ this.recoveryTask = recoveryExecutor.submit(
+ () -> runRecovery(newRecoveryFuture, newRecoveryWorkFinishedFuture));
+ } catch (RejectedExecutionException e) {
+ this.state = State.IDLE;
+ submissionFailure = e;
+ }
+ }
+ if (submissionFailure != null) {
+ newRecoveryWorkFinishedFuture.complete(null);
+ newRecoveryFuture.completeExceptionally(submissionFailure);
+ }
+ // Do not let callers complete the internal recovery result.
+ return newRecoveryFuture.copy();
+ }
+
+ private void runRecovery(CompletableFuture<Position> recoveryResult,
+ CompletableFuture<Void> recoveryWorkFinished) {
+ Position recoveredPosition = null;
+ Throwable recoveryFailure = null;
+ boolean recoveryFinished = false;
+ if (tryStartRecovery()) {
+ try {
+ recoveredPosition = doRecoverFromSnapshot(recoveryExecutor);
+ } catch (Throwable throwable) {
+ recoveryFailure = throwable;
+ }
+ recoveryFinished = tryMarkRecoveryFinished();
+ }
+ recoveryWorkFinished.complete(null);
+ if (!recoveryFinished) {
+ failRecoveryAfterClose(recoveryResult);
+ } else if (recoveryFailure == null) {
+ recoveryResult.complete(recoveredPosition);
+ } else {
+ recoveryResult.completeExceptionally(recoveryFailure);
+ }
+ }
+
+ /**
+ * Recovers the processor state synchronously on the recovery executor thread.
+ * If closure wins while this method is running, its result is discarded and recovery fails as closed.
+ *
+ * @param executor executor used by recovery-related operations
+ * @return the recovery position, or {@code null} when no snapshot exists
+ */
+ abstract Position doRecoverFromSnapshot(ExecutorService executor) throws Exception;
+
+ final boolean isClosed() {
+ return this.state == State.CLOSED;
+ }
+
+ private synchronized boolean tryStartRecovery() {
+ if (this.state != State.RECOVERY_QUEUED) {
+ return false;
+ }
+ this.state = State.RECOVERY_RUNNING;
+ return true;
+ }
+
+ private synchronized boolean tryMarkRecoveryFinished() {
+ if (this.state != State.RECOVERY_RUNNING) {
+ return false;
+ }
+ this.state = State.IDLE;
+ return true;
+ }
+
+ @Override
+ public final CompletableFuture<Void> closeAsync() {
+ State previousState;
+ CompletableFuture<Position> currentRecoveryFuture;
+ CompletableFuture<Void> currentRecoveryWorkFinishedFuture;
+ synchronized (this) {
+ if (this.state == State.CLOSED) {
+ return closeFuture;
+ }
+ previousState = this.state;
+ this.state = State.CLOSED;
+ currentRecoveryFuture = this.recoveryFuture;
+ currentRecoveryWorkFinishedFuture = this.recoveryWorkFinishedFuture;
+ if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) {
+ this.recoveryTask.cancel(false);
+ }
+ }
+ FutureUtil.completeAfter(closeFuture,
+ currentRecoveryWorkFinishedFuture.thenCompose(v -> closeResources()));
+ if (previousState == State.RECOVERY_QUEUED) {
+ failRecoveryAfterClose(currentRecoveryFuture);
+ }
+ if (previousState != State.RECOVERY_RUNNING) {
+ currentRecoveryWorkFinishedFuture.complete(null);
+ }
+ return closeFuture;
+ }
+
+ private void failRecoveryAfterClose(CompletableFuture<Position> recoveryResult) {
+ closeFuture.whenComplete((__, ___) -> recoveryResult.completeExceptionally(closedException()));
+ }
+
+ /**
+ * Closes resources after processor-owned recovery work has finished.
+ *
+ * <p>This method may be invoked on the caller of {@link #closeAsync()} or on the recovery executor thread.
+ * Implementations must not rely on thread affinity and should return without blocking.
+ */
+ abstract CompletableFuture<Void> closeResources();
+
+ private static BrokerServiceException closedException() {
+ return new BrokerServiceException.ServiceUnitNotReadyException(
+ "Transaction buffer snapshot processor is closed");
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
index c737da2ed0e8..9aaab353db82 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.transaction.buffer.impl;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
@@ -29,7 +30,6 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
-import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.metadata.AbortTxnMetadata;
import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.transaction.TxnID;
@@ -39,7 +39,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
@Slf4j
-public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcessor {
+public class SingleSnapshotAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor {
private final PersistentTopic topic;
private final ReferenceCountedWriter<TransactionBufferSnapshot> takeSnapshotWriter;
/**
@@ -50,11 +50,11 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
private volatile long lastSnapshotTimestamps;
- private volatile boolean isClosed = false;
-
public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) {
+ super(topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .getExecutor(TopicName.get(topic.getName()).getNamespace()));
this.topic = topic;
- this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar()
+ this.takeSnapshotWriter = topic.getBrokerService().getPulsar()
.getTransactionBufferSnapshotServiceFactory()
.getTxnBufferSnapshotService().getReferenceWriter(TopicName.get(topic.getName()).getNamespaceObject());
this.takeSnapshotWriter.getFuture().exceptionally((ex) -> {
@@ -88,26 +88,15 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
- final var future = new CompletableFuture<Position>();
+ Position doRecoverFromSnapshot(ExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- pulsar.getTransactionSnapshotRecoverExecutorProvider().getExecutor(this).execute(() -> {
- try {
- final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
- .getTableView().readLatest(topic.getName());
- if (snapshot != null) {
- handleSnapshot(snapshot);
- final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
- snapshot.getMaxReadPositionEntryId());
- future.complete(startReadCursorPosition);
- } else {
- future.complete(null);
- }
- } catch (Throwable e) {
- future.completeExceptionally(e);
- }
- });
- return future;
+ final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
+ .getTableView().readLatest(topic.getName());
+ if (isClosed() || snapshot == null) {
+ return null;
+ }
+ handleSnapshot(snapshot);
+ return PositionFactory.create(snapshot.getMaxReadPositionLedgerId(), snapshot.getMaxReadPositionEntryId());
}
@Override
@@ -167,11 +156,8 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
}
@Override
- public synchronized CompletableFuture<Void> closeAsync() {
- if (!isClosed) {
- isClosed = true;
- takeSnapshotWriter.release();
- }
+ CompletableFuture<Void> closeResources() {
+ takeSnapshotWriter.release();
return CompletableFuture.completedFuture(null);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
index 38b902dc5208..dbda76574a23 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
@@ -28,6 +28,7 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
@@ -48,7 +49,6 @@ import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.Refe
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
-import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex;
import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes;
@@ -71,7 +71,7 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
@Slf4j
-public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcessor {
+public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor {
/**
* Stored the unsealed aborted transaction IDs Whose size is always less than the snapshotSegmentCapacity.
@@ -137,10 +137,14 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
* <p> Clear all snapshot segment. </p>
*/
private final PersistentWorker persistentWorker;
+ // A failed recovery can be retried, so close must retain updates started by every attempt.
+ private CompletableFuture<Void> recoveryIndexUpdatesFuture = CompletableFuture.completedFuture(null);
private static final String SNAPSHOT_PREFIX = "multiple-";
public SnapshotSegmentAbortedTxnProcessorImpl(PersistentTopic topic) {
+ super(topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .getExecutor(TopicName.get(topic.getName()).getNamespace()));
this.topic = topic;
this.persistentWorker = new PersistentWorker(topic);
/*
@@ -228,44 +232,48 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
+ Position doRecoverFromSnapshot(ExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- final var future = new CompletableFuture<Position>();
- pulsar.getTransactionSnapshotRecoverExecutorProvider().getExecutor(this).execute(() -> {
- try {
- final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
- .getTxnBufferSnapshotIndexService().getTableView().readLatest(topic.getName());
- if (indexes == null) {
- // Try recovering from the old format snapshot
- future.complete(recoverOldSnapshot());
- return;
- }
- final var snapshot = indexes.getSnapshot();
- final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
- snapshot.getMaxReadPositionEntryId());
- this.unsealedTxnIds = convertTypeToTxnID(snapshot.getAborts());
- // Read snapshot segment to recover aborts
- final var snapshotSegmentTopicName = TopicName.get(TopicDomain.persistent.toString(),
- TopicName.get(topic.getName()).getNamespaceObject(),
- SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS);
- readSegmentEntries(snapshotSegmentTopicName, indexes);
- if (!this.indexes.isEmpty()) {
- // If there is no segment index, the persistent worker will write segment begin from 0.
- persistentWorker.sequenceID.set(this.indexes.get(this.indexes.lastKey()).sequenceID + 1);
- }
- unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID));
- future.complete(startReadCursorPosition);
- } catch (Throwable throwable) {
- future.completeExceptionally(throwable);
- }
- });
- return future;
+ final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
+ .getTxnBufferSnapshotIndexService().getTableView()
+ .readLatest(topic.getName());
+ if (isClosed()) {
+ return null;
+ }
+ if (indexes == null) {
+ // Try recovering from the old format snapshot
+ return recoverOldSnapshot(executor);
+ }
+ final var snapshot = indexes.getSnapshot();
+ final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
+ snapshot.getMaxReadPositionEntryId());
+ this.unsealedTxnIds = convertTypeToTxnID(snapshot.getAborts());
+ // Read snapshot segment to recover aborts
+ final var snapshotSegmentTopicName = TopicName.get(TopicDomain.persistent.toString(),
+ TopicName.get(topic.getName()).getNamespaceObject(),
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS);
+ readSegmentEntries(snapshotSegmentTopicName, indexes);
+ if (isClosed()) {
+ return null;
+ }
+ if (!this.indexes.isEmpty()) {
+ // If there is no segment index, the persistent worker will write segment begin from 0.
+ persistentWorker.sequenceID.set(this.indexes.get(this.indexes.lastKey()).sequenceID + 1);
+ }
+ unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID));
+ return startReadCursorPosition;
}
private void readSegmentEntries(TopicName topicName, TransactionBufferSnapshotIndexes indexes) throws Exception {
+ if (isClosed()) {
+ return;
+ }
final var managedLedger = openReadOnlyManagedLedger(topicName);
boolean hasInvalidIndex = false;
for (var index : indexes.getIndexList()) {
+ if (isClosed()) {
+ return;
+ }
final var position = PositionFactory.create(index.getSegmentLedgerID(), index.getSegmentEntryID());
final var abortedPosition = PositionFactory.create(index.abortedMarkLedgerID, index.abortedMarkEntryID);
try {
@@ -288,10 +296,11 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
}
}
- if (hasInvalidIndex) {
+ if (hasInvalidIndex && !isClosed()) {
// Update the snapshot segment index if there exist invalid indexes.
- persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex,
- () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot()));
+ recoveryIndexUpdatesFuture = CompletableFuture.allOf(recoveryIndexUpdatesFuture,
+ persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex,
+ () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot())));
}
}
@@ -341,18 +350,24 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
// This method will be deprecated and removed in version 4.x.0
- private Position recoverOldSnapshot() throws Exception {
+ private Position recoverOldSnapshot(ExecutorService executor) throws Exception {
+ if (isClosed()) {
+ return null;
+ }
final var pulsar = topic.getBrokerService().getPulsar();
final var topicName = TopicName.get(topic.getName());
final var topics = wait(pulsar.getPulsarResources().getTopicResources().listPersistentTopicsAsync(
NamespaceName.get(topicName.getNamespace())), "list persistent topics");
+ if (isClosed()) {
+ return null;
+ }
if (!topics.contains(TopicDomain.persistent + "://" + topicName.getNamespace() + "/"
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT)) {
return null;
}
final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
.getTableView().readLatest(topic.getName());
- if (snapshot == null) {
+ if (isClosed() || snapshot == null) {
return null;
}
handleOldSnapshot(snapshot);
@@ -400,8 +415,9 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
@Override
- public CompletableFuture<Void> closeAsync() {
- return persistentWorker.closeAsync();
+ CompletableFuture<Void> closeResources() {
+ return recoveryIndexUpdatesFuture.handle((__, throwable) -> null)
+ .thenCompose(__ -> persistentWorker.closeAsync());
}
private void handleSnapshotSegmentEntry(Entry entry) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 385396f3727c..8046104ba627 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -26,12 +26,13 @@ import io.netty.util.TimerTask;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
-import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
@@ -52,6 +53,7 @@ import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSna
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
import org.apache.pulsar.common.protocol.Commands;
@@ -90,7 +92,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final int takeSnapshotIntervalTime;
-
+ // Complete outside the buffer monitor: inline continuations may acquire the topic monitor.
private final CompletableFuture<Void> transactionBufferFuture = new CompletableFuture<>();
/**
@@ -106,6 +108,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final AbortedTxnProcessor.SnapshotType snapshotType;
private final MaxReadPositionCallBack maxReadPositionCallBack;
+ private Future<?> recoveryReplayTask;
/** if the first snapshot is in progress, it will pending following publishing tasks. **/
private final LinkedList<PendingAppendingTxnBufferTask> pendingAppendingTxnBufferTasks = new LinkedList<>();
@@ -148,37 +151,47 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
.execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() {
@Override
public void recoverComplete() {
+ BrokerServiceException recoveryFailure = null;
synchronized (TopicTransactionBuffer.this) {
+ if (checkIfClosed()) {
+ return;
+ }
if (ongoingTxns.isEmpty()) {
updateMaxReadPositionAfterRecovery();
}
if (!changeToReadyState()) {
log.error("[{}]Transaction buffer recover fail, current state: {}",
topic.getName(), getState());
- getTransactionBufferFuture().completeExceptionally
- (new BrokerServiceException.ServiceUnitNotReadyException(
- "Transaction buffer recover failed to change the status to Ready,"
- + "current state is: " + getState()));
+ recoveryFailure = new BrokerServiceException.ServiceUnitNotReadyException(
+ "Transaction buffer recover failed to change the status to Ready,"
+ + "current state is: " + getState());
} else {
timer.newTimeout(TopicTransactionBuffer.this,
takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
- getTransactionBufferFuture().complete(null);
recoverTime.setRecoverEndTime(System.currentTimeMillis());
}
}
+ if (recoveryFailure != null) {
+ getTransactionBufferFuture().completeExceptionally(recoveryFailure);
+ } else {
+ getTransactionBufferFuture().complete(null);
+ }
}
@Override
public void noNeedToRecover() {
synchronized (TopicTransactionBuffer.this) {
+ if (checkIfClosed()) {
+ return;
+ }
updateMaxReadPositionAfterRecovery();
if (!changeToNoSnapshotState()) {
log.error("[{}]Transaction buffer recover fail", topic.getName());
- } else {
- getTransactionBufferFuture().complete(null);
- recoverTime.setRecoverEndTime(System.currentTimeMillis());
+ return;
}
+ recoverTime.setRecoverEndTime(System.currentTimeMillis());
}
+ getTransactionBufferFuture().complete(null);
}
@Override
public void handleTxnEntry(Entry entry) {
@@ -192,8 +205,12 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
if (msgMetadata != null && msgMetadata.hasTxnidMostBits() && msgMetadata.hasTxnidLeastBits()) {
TxnID txnID = new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits());
Position position = PositionFactory.create(entry.getLedgerId(), entry.getEntryId());
+ boolean isTxnMarker = Markers.isTxnMarker(msgMetadata);
synchronized (TopicTransactionBuffer.this) {
- if (Markers.isTxnMarker(msgMetadata)) {
+ if (checkIfClosed()) {
+ return;
+ }
+ if (isTxnMarker) {
if (Markers.isTxnAbortMarker(msgMetadata)) {
snapshotAbortedTxnProcessor.putAbortedTxnAndPosition(txnID, position);
}
@@ -202,14 +219,14 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
handleTransactionMessage(txnID, position);
}
}
+ if (isTxnMarker) {
+ updateLastDispatchablePosition(null);
+ }
}
}
@Override
public void recoverExceptionally(Throwable e) {
-
- log.warn("Closing topic {} due to read transaction buffer snapshot while recovering the "
- + "transaction buffer throw exception", topic.getName(), e);
// when create reader or writer fail throw PulsarClientException,
// should close this topic and then reinit this topic
if (e instanceof PulsarClientException) {
@@ -222,11 +239,25 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
getTransactionBufferFuture().completeExceptionally(e);
}
recoverTime.setRecoverEndTime(System.currentTimeMillis());
+ if (checkIfClosed() || topic.isClosingOrDeleting()) {
+ return;
+ }
+ log.warn("Closing topic {} due to read transaction buffer snapshot while recovering the "
+ + "transaction buffer throw exception", topic.getName(), e);
topic.close(true);
}
}, this.topic, this, snapshotAbortedTxnProcessor));
}
+ private synchronized void submitRecoveryReplay(Runnable replay) {
+ if (checkIfClosed()) {
+ return;
+ }
+ // Replay polls for ledger reads, so it must stay off the live transaction executor.
+ recoveryReplayTask = topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .getExecutor(TopicName.get(topic.getName()).getNamespace()).submit(replay);
+ }
+
@Override
public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
return CompletableFuture.completedFuture(null);
@@ -309,6 +340,11 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
PendingAppendingTxnBufferTask pendingTask = null;
try {
synchronized (pendingAppendingTxnBufferTasks) {
+ if (checkIfClosed()) {
+ failPendingTasks.accept(
+ new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed"));
+ return;
+ }
while ((pendingTask = pendingAppendingTxnBufferTasks.poll()) != null) {
final ByteBuf data = pendingTask.buffer;
final CompletableFuture<Position> pendingFuture =
@@ -434,7 +470,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
- // ThreadSafe
+ // Acquires the topic monitor: never call while holding the buffer monitor, since topic close takes both.
private void updateLastDispatchablePosition(Position position) {
topic.updateLastDispatchablePosition(position);
}
@@ -464,6 +500,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
snapshotAbortedTxnProcessor.trimExpiredAbortedTxns();
takeSnapshotByChangeTimes();
}
+ updateLastDispatchablePosition(null);
txnCommittedCounter.increment();
completableFuture.complete(null);
}
@@ -509,8 +546,13 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
removeTxnAndUpdateMaxReadPosition(txnID);
snapshotAbortedTxnProcessor.trimExpiredAbortedTxns();
takeSnapshotByChangeTimes();
- txnAbortedCounter.increment();
- completableFuture.complete(null);
+ }
+ updateLastDispatchablePosition(null);
+ txnAbortedCounter.increment();
+ completableFuture.complete(null);
+ // Complete first so the low-water-mark abort continuation releases its permit before
+ // we scan for the next transaction. Completion must remain outside the buffer monitor.
+ synchronized (TopicTransactionBuffer.this) {
handleLowWaterMark(txnID, lowWaterMark);
}
}
@@ -601,8 +643,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
} else {
updateMaxReadPosition(topic.getManagedLedger().getLastConfirmedEntry(), false);
}
- // Update the last dispatchable position to null if there is a TXN finished.
- updateLastDispatchablePosition(null);
}
/**
@@ -679,17 +719,30 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
@Override
public CompletableFuture<Void> closeAsync() {
- synchronized (pendingAppendingTxnBufferTasks) {
- if (!checkIfClosed()) {
+ boolean closeStarted;
+ // Serialize closure with recovery entry handling and replay submission, which use the same monitor.
+ synchronized (this) {
+ closeStarted = !checkIfClosed();
+ changeToCloseState();
+ if (recoveryReplayTask != null) {
+ recoveryReplayTask.cancel(false);
+ recoveryReplayTask = null;
+ }
+ }
+ // Cancel snapshot recovery before completing futures whose callbacks may run inline.
+ CompletableFuture<Void> processorCloseFuture = this.snapshotAbortedTxnProcessor.closeAsync();
+ if (closeStarted) {
+ Throwable closeException =
+ new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed");
+ getTransactionBufferFuture().completeExceptionally(closeException);
+ synchronized (pendingAppendingTxnBufferTasks) {
PendingAppendingTxnBufferTask pendingTask = null;
- Throwable t = new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed");
while ((pendingTask = pendingAppendingTxnBufferTasks.poll()) != null) {
- pendingTask.fail(t);
+ pendingTask.fail(closeException);
}
}
- changeToCloseState();
}
- return this.snapshotAbortedTxnProcessor.closeAsync();
+ return processorCloseFuture;
}
@Override
@@ -793,8 +846,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final TopicTransactionBufferRecoverCallBack callBack;
- private Position startReadCursorPosition = PositionFactory.EARLIEST;
-
private final SpscArrayQueue<Entry> entryQueue;
private final AtomicLong exceptionNumber = new AtomicLong();
@@ -815,7 +866,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
this.abortedTxnProcessor = abortedTxnProcessor;
}
- @SneakyThrows
@Override
public void run() {
if (!this.topicTransactionBuffer.changeToInitializingState()) {
@@ -823,30 +873,62 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
this, topic.getName());
return;
}
- abortedTxnProcessor.recoverFromSnapshot().thenAccept(startReadCursorPosition -> {
- //Transaction is not use for this topic, so just make maxReadPosition as LAC.
- if (startReadCursorPosition == null) {
- callBack.noNeedToRecover();
- return;
- } else {
- this.startReadCursorPosition = startReadCursorPosition;
- }
- ManagedCursor managedCursor;
+ abortedTxnProcessor.recoverFromSnapshot().thenAccept(this::submitReplay)
+ .exceptionally(this::handleRecoveryFailure);
+ }
+
+ private void submitReplay(Position recoveredPosition) {
+ topicTransactionBuffer.submitRecoveryReplay(() -> {
try {
- managedCursor = topic.getManagedLedger()
- .newNonDurableCursor(this.startReadCursorPosition, SUBSCRIPTION_NAME);
- } catch (ManagedLedgerException e) {
+ replayTransactionBuffer(recoveredPosition);
+ } catch (Throwable error) {
+ handleRecoveryFailure(error);
+ }
+ });
+ }
+
+ private Void handleRecoveryFailure(Throwable error) {
+ Throwable cause = FutureUtil.unwrapCompletionException(error);
+ if (!shouldStopRecovery()) {
+ log.error("[{}]Transaction buffer failed to recover snapshot!", topic.getName(), cause);
+ }
+ callBack.recoverExceptionally(cause);
+ return null;
+ }
+
+ private boolean shouldStopRecovery() {
+ // Topic deletion can be rolled back; only buffer closure is irreversible.
+ return topicTransactionBuffer.checkIfClosed();
+ }
+
+ private void replayTransactionBuffer(Position recoveredPosition) {
+ if (shouldStopRecovery()) {
+ return;
+ }
+ // Transaction is not used for this topic, so just make maxReadPosition as LAC.
+ if (recoveredPosition == null) {
+ callBack.noNeedToRecover();
+ return;
+ }
+ ManagedCursor managedCursor;
+ try {
+ managedCursor = topic.getManagedLedger()
+ .newNonDurableCursor(recoveredPosition, SUBSCRIPTION_NAME);
+ } catch (ManagedLedgerException e) {
+ if (!shouldStopRecovery()) {
callBack.recoverExceptionally(e);
log.error("[{}]Transaction buffer recover fail when open cursor!", topic.getName(), e);
- return;
}
- Position lastConfirmedEntry =
- topic.getManagedLedger().getLastConfirmedEntry();
- Position currentLoadPosition = this.startReadCursorPosition;
- FillEntryQueueCallback fillEntryQueueCallback = new FillEntryQueueCallback(entryQueue,
- managedCursor, TopicTransactionBufferRecover.this);
+ return;
+ }
+ Position lastConfirmedEntry = topic.getManagedLedger().getLastConfirmedEntry();
+ Position currentLoadPosition = recoveredPosition;
+ FillEntryQueueCallback fillEntryQueueCallback = new FillEntryQueueCallback(entryQueue,
+ managedCursor, TopicTransactionBufferRecover.this);
+ try {
if (lastConfirmedEntry.getEntryId() != -1) {
- while (lastConfirmedEntry.compareTo(currentLoadPosition) > 0
+ while (!shouldStopRecovery()
+ && lastConfirmedEntry.compareTo(currentLoadPosition) > 0
&& fillEntryQueueCallback.fillQueue()) {
Entry entry = entryQueue.poll();
if (entry != null) {
@@ -861,19 +943,19 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
try {
Thread.sleep(1);
} catch (InterruptedException e) {
- //no-op
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
}
}
}
}
-
+ } finally {
+ fillEntryQueueCallback.stopAndReleasePendingEntries();
closeCursor(SUBSCRIPTION_NAME);
+ }
+ if (!shouldStopRecovery()) {
callBack.recoverComplete();
- }).exceptionally(e -> {
- callBack.recoverExceptionally(e.getCause());
- log.error("[{}]Transaction buffer failed to recover snapshot!", topic.getName(), e);
- return null;
- });
+ }
}
private void closeCursor(String subscriptionName) {
@@ -929,6 +1011,8 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private volatile boolean isReadable = true;
+ private boolean stopped;
+
private static final int NUMBER_OF_PER_READ_ENTRY = 100;
private FillEntryQueueCallback(SpscArrayQueue<Entry> entryQueue, ManagedCursor cursor,
@@ -954,7 +1038,12 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
@Override
- public void readEntriesComplete(List<Entry> entries, Object ctx) {
+ public synchronized void readEntriesComplete(List<Entry> entries, Object ctx) {
+ if (stopped) {
+ // An asynchronous read can complete after replay has stopped and no longer has a consumer.
+ entries.forEach(Entry::release);
+ return;
+ }
entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
private int i = 0;
@Override
@@ -968,8 +1057,19 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
outstandingReadsRequests.decrementAndGet();
}
+ private synchronized void stopAndReleasePendingEntries() {
+ stopped = true;
+ Entry entry;
+ while ((entry = entryQueue.poll()) != null) {
+ entry.release();
+ }
+ }
+
@Override
- public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ public synchronized void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ if (stopped) {
+ return;
+ }
if (recover.topic.getManagedLedger().getConfig().isAutoSkipNonRecoverableData()
&& exception instanceof ManagedLedgerException.NonRecoverableLedgerException
|| exception instanceof ManagedLedgerException.ManagedLedgerFencedException
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
index e2317d4a8279..527619297298 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -38,6 +38,7 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import io.netty.buffer.Unpooled;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
@@ -1648,7 +1649,7 @@ public class TransactionTest extends TransactionTestBase {
public void testTBRecoverChangeStateError() throws InterruptedException, TimeoutException {
final AtomicReference<PersistentTopic> persistentTopic = new AtomicReference<>();
// Create Executor
- ScheduledExecutorService executorServiceRecover = mock(ScheduledExecutorService.class);
+ ListeningScheduledExecutorService executorServiceRecover = mock(ListeningScheduledExecutorService.class);
// Mock serviceConfiguration.
ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class);
when(serviceConfiguration.isEnableReplicatedSubscriptions()).thenReturn(false);
@@ -1702,6 +1703,9 @@ public class TransactionTest extends TransactionTestBase {
when(pulsar.getConfiguration()).thenReturn(serviceConfiguration);
when(pulsar.getConfig()).thenReturn(serviceConfiguration);
when(pulsar.getTransactionExecutorProvider()).thenReturn(executorProvider);
+ ExecutorProvider snapshotRecoverExecutorProvider = mock(ExecutorProvider.class);
+ when(snapshotRecoverExecutorProvider.getExecutor(any(Object.class))).thenReturn(executorServiceRecover);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(snapshotRecoverExecutorProvider);
when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(transactionBufferSnapshotServiceFactory);
TopicTransactionBufferProvider topicTransactionBufferProvider = new TopicTransactionBufferProvider();
when(pulsar.getTransactionBufferProvider()).thenReturn(topicTransactionBufferProvider);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java
new file mode 100644
index 000000000000..ad456f85dadf
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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.transaction.buffer.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.policies.data.TransactionBufferStats;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class AbstractSnapshotAbortedTxnProcessorTest {
+
+ @Test(timeOut = 10_000)
+ public void testCloseCancelsQueuedRecovery() throws Exception {
+ try (RecoveryTestContext context = RecoveryTestContext.queued()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ CompletableFuture<Boolean> callbackHeldProcessorLock =
+ trackCallbackLock(recoveryFuture, context.processor);
+ CompletableFuture<Boolean> closeCompletedBeforeRecovery = recoveryFuture.handle(
+ (__, ___) -> context.processor.closeAsync().isDone());
+ CompletableFuture<?> repeatedRecoveryFuture = context.processor.recoverFromSnapshot();
+ context.awaitRecoveryQueued();
+
+ context.processor.closeAsync().get(5, TimeUnit.SECONDS);
+
+ assertTrue(context.submittedRecoveryTask().isCancelled());
+ context.verifyRecoveryFailedAfterClose(recoveryFuture);
+ context.verifyRecoveryFailedAfterClose(repeatedRecoveryFuture);
+ assertFalse(context.processor.recoveryStarted());
+ assertTrue(context.processor.resourcesClosed());
+ assertFalse(callbackHeldProcessorLock.get(1, TimeUnit.SECONDS),
+ "Recovery callbacks must run without holding the processor lock");
+ assertTrue(closeCompletedBeforeRecovery.get(1, TimeUnit.SECONDS),
+ "Close must finish before publishing a close-induced recovery failure");
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseWaitsForRunningRecovery() throws Exception {
+ try (RecoveryTestContext context = RecoveryTestContext.running()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ context.awaitRecoveryStarted();
+
+ Future<?> recoveryTask = context.submittedRecoveryTask();
+ CompletableFuture<Void> closeFuture = context.processor.closeAsync();
+
+ assertFalse(recoveryTask.isCancelled());
+ assertFalse(recoveryFuture.isDone(), "Running recovery must not complete until its task stops");
+ assertFalse(closeFuture.isDone(), "Closing must wait for running recovery before closing resources");
+ assertFalse(context.processor.resourcesClosed());
+
+ context.finishRecovery();
+ context.verifyRecoveryFailedAfterClose(recoveryFuture);
+ closeFuture.get(5, TimeUnit.SECONDS);
+ assertTrue(context.processor.resourcesClosed());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseDoesNotWaitForRecoveryCallback() throws Exception {
+ CountDownLatch callbackStarted = new CountDownLatch(1);
+ CountDownLatch finishCallback = new CountDownLatch(1);
+ try (RecoveryTestContext context = RecoveryTestContext.running()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ CompletableFuture<Void> callbackFuture = recoveryFuture.thenRun(() -> {
+ assertFalse(Thread.holdsLock(context.processor),
+ "Recovery callbacks must run without holding the processor lock");
+ callbackStarted.countDown();
+ try {
+ assertTrue(finishCallback.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ });
+ context.awaitRecoveryStarted();
+
+ context.finishRecovery();
+ assertTrue(callbackStarted.await(5, TimeUnit.SECONDS));
+ CompletableFuture<Void> closeFuture = context.processor.closeAsync();
+
+ closeFuture.get(5, TimeUnit.SECONDS);
+ assertTrue(context.processor.resourcesClosed());
+ context.verifyRecoveryFailedAfterClose(context.processor.recoverFromSnapshot());
+
+ finishCallback.countDown();
+ callbackFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ finishCallback.countDown();
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testRecoverySubmissionRejected() throws Exception {
+ TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor();
+ recoveryExecutor.shutdown();
+ TestSnapshotProcessor processor = new TestSnapshotProcessor(recoveryExecutor);
+
+ CompletableFuture<Position> recoveryFuture = processor.recoverFromSnapshot();
+
+ ExecutionException exception = expectThrows(ExecutionException.class, recoveryFuture::get);
+ assertTrue(exception.getCause() instanceof RejectedExecutionException);
+ processor.closeAsync().get(5, TimeUnit.SECONDS);
+ assertTrue(processor.resourcesClosed());
+ }
+
+ private static CompletableFuture<Boolean> trackCallbackLock(CompletableFuture<?> recoveryFuture,
+ Object processor) {
+ CompletableFuture<Boolean> callbackHeldProcessorLock = new CompletableFuture<>();
+ recoveryFuture.whenComplete((__, ___) ->
+ callbackHeldProcessorLock.complete(Thread.holdsLock(processor)));
+ return callbackHeldProcessorLock;
+ }
+
+ private static final class RecoveryTestContext implements AutoCloseable {
+
+ private final TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor();
+ private final CountDownLatch releaseBlocker = new CountDownLatch(1);
+ private final TestSnapshotProcessor processor;
+
+ static RecoveryTestContext queued() throws Exception {
+ return new RecoveryTestContext(true);
+ }
+
+ static RecoveryTestContext running() throws Exception {
+ return new RecoveryTestContext(false);
+ }
+
+ private RecoveryTestContext(boolean queueRecovery) throws Exception {
+ if (queueRecovery) {
+ CountDownLatch blockerStarted = new CountDownLatch(1);
+ recoveryExecutor.execute(() -> {
+ blockerStarted.countDown();
+ try {
+ releaseBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(blockerStarted.await(5, TimeUnit.SECONDS));
+ }
+
+ processor = new TestSnapshotProcessor(recoveryExecutor);
+ }
+
+ void awaitRecoveryQueued() {
+ assertEquals(recoveryExecutor.getQueue().size(), 1);
+ assertFalse(submittedRecoveryTask().isCancelled());
+ }
+
+ void awaitRecoveryStarted() throws InterruptedException {
+ assertTrue(processor.awaitRecoveryStarted());
+ }
+
+ Future<?> submittedRecoveryTask() {
+ Future<?> recoveryTask = recoveryExecutor.submittedTask();
+ assertNotNull(recoveryTask);
+ return recoveryTask;
+ }
+
+ void finishRecovery() {
+ processor.finishRecovery();
+ }
+
+ void verifyRecoveryFailedAfterClose(CompletableFuture<?> recoveryFuture) throws Exception {
+ ExecutionException exception = expectThrows(ExecutionException.class,
+ () -> recoveryFuture.get(5, TimeUnit.SECONDS));
+ assertTrue(exception.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException,
+ "Closing the processor must fail the recovery future");
+ }
+
+ @Override
+ public void close() throws Exception {
+ releaseBlocker.countDown();
+ processor.finishRecovery();
+ recoveryExecutor.shutdownNow();
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ private static final class TrackingScheduledExecutor extends ScheduledThreadPoolExecutor {
+
+ private final AtomicReference<Future<?>> submittedTask = new AtomicReference<>();
+
+ private TrackingScheduledExecutor() {
+ super(1);
+ }
+
+ @Override
+ public Future<?> submit(Runnable task) {
+ Future<?> future = super.submit(task);
+ submittedTask.set(future);
+ return future;
+ }
+
+ private Future<?> submittedTask() {
+ return submittedTask.get();
+ }
+ }
+
+ private static final class TestSnapshotProcessor extends AbstractSnapshotAbortedTxnProcessor {
+
+ private final CountDownLatch recoveryStarted = new CountDownLatch(1);
+ private final CountDownLatch finishRecovery = new CountDownLatch(1);
+ private final AtomicBoolean resourcesClosed = new AtomicBoolean();
+
+ private TestSnapshotProcessor(ExecutorService recoveryExecutor) {
+ super(recoveryExecutor);
+ }
+
+ @Override
+ Position doRecoverFromSnapshot(ExecutorService executor) throws Exception {
+ recoveryStarted.countDown();
+ assertTrue(finishRecovery.await(5, TimeUnit.SECONDS));
+ return null;
+ }
+
+ @Override
+ CompletableFuture<Void> closeResources() {
+ resourcesClosed.set(true);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void putAbortedTxnAndPosition(TxnID txnID, Position position) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void trimExpiredAbortedTxns() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean checkAbortedTransaction(TxnID txnID) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> clearAbortedTxnSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> takeAbortedTxnsSnapshot(Position maxReadPosition) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) {
+ throw new UnsupportedOperationException();
+ }
+
+ boolean awaitRecoveryStarted() throws InterruptedException {
+ return recoveryStarted.await(5, TimeUnit.SECONDS);
+ }
+
+ boolean recoveryStarted() {
+ return recoveryStarted.getCount() == 0;
+ }
+
+ void finishRecovery() {
+ finishRecovery.countDown();
+ }
+
+ boolean resourcesClosed() {
+ return resourcesClosed.get();
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java
new file mode 100644
index 000000000000..37de7244f37a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.transaction.buffer.impl;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.Collections;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.bookkeeper.mledger.ReadOnlyManagedLedger;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService;
+import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter;
+import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceFactory;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexesMetadata;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotSegment;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class SnapshotSegmentAbortedTxnProcessorCloseTest {
+
+ @Test(timeOut = 10_000)
+ @SuppressWarnings("unchecked")
+ public void testCloseWaitsForRecoveryIndexUpdate() throws Exception {
+ ListeningScheduledExecutorService recoveryExecutor =
+ MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor());
+ CompletableFuture<MessageId> indexUpdateFuture = new CompletableFuture<>();
+ CountDownLatch indexUpdateStarted = new CountDownLatch(1);
+ SnapshotSegmentAbortedTxnProcessorImpl processor = null;
+ try {
+ String topicName = "persistent://public/default/test-close-during-index-update";
+ PersistentTopic topic = mock(PersistentTopic.class);
+ BrokerService brokerService = mock(BrokerService.class);
+ PulsarService pulsar = mock(PulsarService.class);
+ ServiceConfiguration configuration = mock(ServiceConfiguration.class);
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ ClientConfigurationData clientConfiguration = new ClientConfigurationData();
+ clientConfiguration.setOperationTimeoutMs(5_000);
+ ExecutorProvider recoveryScheduler = mock(ExecutorProvider.class);
+ TransactionBufferSnapshotServiceFactory serviceFactory =
+ mock(TransactionBufferSnapshotServiceFactory.class);
+
+ when(topic.getName()).thenReturn(topicName);
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(pulsar.getConfiguration()).thenReturn(configuration);
+ when(pulsar.getClient()).thenReturn(client);
+ when(client.getConfiguration()).thenReturn(clientConfiguration);
+ when(configuration.getTransactionBufferSnapshotSegmentSize()).thenReturn(1024);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(recoveryScheduler);
+ when(recoveryScheduler.getExecutor(any(Object.class))).thenReturn(recoveryExecutor);
+ when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(serviceFactory);
+
+ SystemTopicTxnBufferSnapshotService<TransactionBufferSnapshotSegment> segmentService =
+ mock(SystemTopicTxnBufferSnapshotService.class);
+ SystemTopicTxnBufferSnapshotService<TransactionBufferSnapshotIndexes> indexService =
+ mock(SystemTopicTxnBufferSnapshotService.class);
+ ReferenceCountedWriter<TransactionBufferSnapshotSegment> segmentWriter =
+ mock(ReferenceCountedWriter.class);
+ ReferenceCountedWriter<TransactionBufferSnapshotIndexes> indexWriter =
+ mock(ReferenceCountedWriter.class);
+ SystemTopicClient.Writer<TransactionBufferSnapshotSegment> segmentSystemWriter =
+ mock(SystemTopicClient.Writer.class);
+ SystemTopicClient.Writer<TransactionBufferSnapshotIndexes> indexSystemWriter =
+ mock(SystemTopicClient.Writer.class);
+
+ when(serviceFactory.getTxnBufferSnapshotSegmentService()).thenReturn(segmentService);
+ when(serviceFactory.getTxnBufferSnapshotIndexService()).thenReturn(indexService);
+ when(segmentService.getReferenceWriter(any())).thenReturn(segmentWriter);
+ when(indexService.getReferenceWriter(any())).thenReturn(indexWriter);
+ when(segmentWriter.getFuture()).thenReturn(CompletableFuture.completedFuture(segmentSystemWriter));
+ when(indexWriter.getFuture()).thenReturn(CompletableFuture.completedFuture(indexSystemWriter));
+ when(indexSystemWriter.writeAsync(anyString(), any())).thenAnswer(invocation -> {
+ indexUpdateStarted.countDown();
+ return indexUpdateFuture;
+ });
+
+ TransactionBufferSnapshotIndex invalidIndex = new TransactionBufferSnapshotIndex(0, 1, 1, 2, 2);
+ TransactionBufferSnapshotIndexesMetadata metadata =
+ new TransactionBufferSnapshotIndexesMetadata(3, 3, Collections.emptyList());
+ TransactionBufferSnapshotIndexes indexes =
+ new TransactionBufferSnapshotIndexes(topicName, Collections.singletonList(invalidIndex), metadata);
+ TableView<TransactionBufferSnapshotIndexes> tableView = mock(TableView.class);
+ when(indexService.getTableView()).thenReturn(tableView);
+ when(tableView.readLatest(topicName)).thenReturn(indexes);
+
+ ManagedLedgerImpl managedLedger = mock(ManagedLedgerImpl.class);
+ when(topic.getManagedLedger()).thenReturn(managedLedger);
+ when(managedLedger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(managedLedger.getLedgersInfo()).thenReturn(new TreeMap<>());
+ ManagedLedgerFactory managedLedgerFactory = mock(ManagedLedgerFactory.class);
+ ReadOnlyManagedLedger readOnlyManagedLedger = mock(ReadOnlyManagedLedger.class);
+ when(brokerService.getManagedLedgerFactoryForTopic(any()))
+ .thenReturn(CompletableFuture.completedFuture(managedLedgerFactory));
+ doAnswer(invocation -> {
+ AsyncCallbacks.OpenReadOnlyManagedLedgerCallback callback = invocation.getArgument(1);
+ callback.openReadOnlyManagedLedgerComplete(readOnlyManagedLedger, invocation.getArgument(3));
+ return null;
+ }).when(managedLedgerFactory).asyncOpenReadOnlyManagedLedger(anyString(), any(), any(), isNull());
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntryCallback callback = invocation.getArgument(1);
+ callback.readEntryFailed(new ManagedLedgerException("missing segment"), invocation.getArgument(2));
+ return null;
+ }).when(readOnlyManagedLedger).asyncReadEntry(any(), any(), isNull());
+
+ processor = new SnapshotSegmentAbortedTxnProcessorImpl(topic);
+ processor.recoverFromSnapshot().get(5, TimeUnit.SECONDS);
+ assertTrue(indexUpdateStarted.await(5, TimeUnit.SECONDS));
+ // A later recovery must not replace the in-flight update in the close barrier.
+ processor.recoverFromSnapshot().get(5, TimeUnit.SECONDS);
+
+ CompletableFuture<Void> closeFuture = processor.closeAsync();
+
+ assertFalse(closeFuture.isDone(), "Closing must wait for the recovery index update");
+ verify(indexWriter, never()).release();
+
+ indexUpdateFuture.complete(MessageId.earliest);
+ closeFuture.get(5, TimeUnit.SECONDS);
+ verify(indexWriter).release();
+ } finally {
+ indexUpdateFuture.complete(MessageId.earliest);
+ if (processor != null) {
+ processor.closeAsync().get(5, TimeUnit.SECONDS);
+ }
+ recoveryExecutor.shutdownNow();
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
new file mode 100644
index 000000000000..c99105a1b956
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
@@ -0,0 +1,276 @@
+/*
+ * 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.transaction.buffer.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.HashedWheelTimer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.MarkerType;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class TopicTransactionBufferCloseTest {
+
+ @DataProvider
+ public Object[][] transactionCompletionModes() {
+ return new Object[][] { { true, false }, { true, true }, { false, false }, { false, true } };
+ }
+
+ @Test(dataProvider = "transactionCompletionModes", timeOut = 10_000)
+ public void testTxnCompletionUpdatesTopicOutsideBufferLock(boolean replay, boolean abort) throws Exception {
+ Position position = PositionFactory.create(1, 1);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, position)) {
+ // Fenced publish failures close the buffer while holding the topic monitor.
+ doAnswer(__ -> {
+ assertThat(Thread.holdsLock(context.transactionBuffer))
+ .as("updating the topic must not acquire its monitor while holding the buffer monitor")
+ .isFalse();
+ return null;
+ }).when(context.topic).updateLastDispatchablePosition(null);
+
+ TxnID txnID = new TxnID(1, 1);
+ if (replay) {
+ Entry marker = mock(Entry.class);
+ when(marker.getLedgerId()).thenReturn(position.getLedgerId());
+ when(marker.getEntryId()).thenReturn(position.getEntryId());
+ when(marker.getMessageMetadata()).thenReturn(new MessageMetadata()
+ .setTxnidMostBits(txnID.getMostSigBits())
+ .setTxnidLeastBits(txnID.getLeastSigBits())
+ .setMarkerType((abort ? MarkerType.TXN_ABORT : MarkerType.TXN_COMMIT).getValue()));
+ when(context.managedCursor.hasMoreEntries()).thenReturn(true, false);
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1);
+ callback.readEntriesComplete(List.of(marker), null);
+ return null;
+ }).when(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+
+ recoveryFuture.complete(PositionFactory.create(1, 0));
+ context.transactionBuffer.getTransactionBufferFuture().get(5, TimeUnit.SECONDS);
+ verify(marker).release();
+ } else {
+ recoveryFuture.complete(position);
+ context.transactionBuffer.getTransactionBufferFuture().get(5, TimeUnit.SECONDS);
+ doAnswer(invocation -> {
+ AsyncCallbacks.AddEntryCallback callback = invocation.getArgument(1);
+ callback.addComplete(position, invocation.getArgument(0), null);
+ return null;
+ }).when(context.managedLedger).asyncAddEntry(any(ByteBuf.class), any(), any());
+
+ CompletableFuture<Void> completion = abort
+ ? context.transactionBuffer.abortTxn(txnID, 0)
+ : context.transactionBuffer.commitTxn(txnID, 0);
+ completion.get(5, TimeUnit.SECONDS);
+ }
+ verify(context.topic).updateLastDispatchablePosition(null);
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseInducedRecoveryFailureDoesNotCloseTopicAgain() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ context.awaitExecutorsIdle();
+ when(context.processor.closeAsync()).thenAnswer(__ -> {
+ recoveryFuture.completeExceptionally(
+ new BrokerServiceException.ServiceUnitNotReadyException("processor closed"));
+ return CompletableFuture.completedFuture(null);
+ });
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ context.awaitExecutorsIdle();
+
+ verify(context.topic, never()).close(true);
+ assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testRecoveryContinuationDoesNotStartAfterClose() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ context.awaitExecutorsIdle();
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ recoveryFuture.complete(PositionFactory.EARLIEST);
+ context.awaitExecutorsIdle();
+
+ verify(context.managedLedger, never()).newNonDurableCursor(any(), anyString());
+ assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseCancelsQueuedRecoveryReplay() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CountDownLatch blockerStarted = new CountDownLatch(1);
+ CountDownLatch releaseBlocker = new CountDownLatch(1);
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ try {
+ context.awaitExecutorsIdle();
+ AtomicReference<Future<?>> submittedTask = new AtomicReference<>();
+ doAnswer(invocation -> {
+ Future<?> task = (Future<?>) invocation.callRealMethod();
+ submittedTask.set(task);
+ return task;
+ }).when(context.recoveryExecutor).submit(any(Runnable.class));
+ context.recoveryExecutor.execute(() -> {
+ blockerStarted.countDown();
+ try {
+ releaseBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(blockerStarted.await(5, TimeUnit.SECONDS));
+
+ recoveryFuture.complete(PositionFactory.EARLIEST);
+ Future<?> replayTask = submittedTask.get();
+ assertNotNull(replayTask);
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+
+ assertTrue(replayTask.isCancelled());
+ } finally {
+ releaseBlocker.countDown();
+ }
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testLateRecoveryReadIsReleasedAfterClose() throws Exception {
+ Position startPosition = PositionFactory.create(1, 0);
+ Position lastPosition = PositionFactory.create(1, 3);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CountDownLatch readStarted = new CountDownLatch(1);
+ AtomicReference<AsyncCallbacks.ReadEntriesCallback> readCallback = new AtomicReference<>();
+ try (TestContext context = new TestContext(recoveryFuture, lastPosition)) {
+ when(context.managedCursor.hasMoreEntries()).thenReturn(true);
+ doAnswer(invocation -> {
+ readCallback.set(invocation.getArgument(1));
+ readStarted.countDown();
+ return null;
+ }).when(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+
+ recoveryFuture.complete(startPosition);
+ assertTrue(readStarted.await(5, TimeUnit.SECONDS));
+ // Live transaction work must still run while the recovery read is outstanding.
+ context.transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ context.awaitExecutorsIdle();
+
+ Entry lateEntry = mock(Entry.class);
+ readCallback.get().readEntriesComplete(List.of(lateEntry), null);
+
+ verify(lateEntry).release();
+ verify(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+ }
+ }
+
+ private static final class TestContext implements AutoCloseable {
+ private final ExecutorService transactionExecutor = Executors.newSingleThreadExecutor();
+ private final ListeningScheduledExecutorService recoveryExecutor =
+ spy(MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor()));
+ private final AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ private final PersistentTopic topic = mock(PersistentTopic.class);
+ private final ManagedLedgerImpl managedLedger = mock(ManagedLedgerImpl.class);
+ private final ManagedCursor managedCursor = mock(ManagedCursor.class);
+ private final TopicTransactionBuffer transactionBuffer;
+
+ private TestContext(CompletableFuture<Position> recoveryFuture, Position lastConfirmedEntry) throws Exception {
+ BrokerService brokerService = mock(BrokerService.class);
+ PulsarService pulsar = mock(PulsarService.class);
+ ServiceConfiguration configuration = mock(ServiceConfiguration.class);
+ ExecutorProvider executorProvider = mock(ExecutorProvider.class);
+ ExecutorProvider recoveryScheduler = mock(ExecutorProvider.class);
+
+ when(topic.getName()).thenReturn("persistent://public/default/test-close-during-recovery");
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(topic.getManagedLedger()).thenReturn(managedLedger);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(pulsar.getConfiguration()).thenReturn(configuration);
+ when(pulsar.getTransactionTimer()).thenReturn(mock(HashedWheelTimer.class));
+ when(pulsar.getTransactionExecutorProvider()).thenReturn(executorProvider);
+ when(executorProvider.getExecutor(any(Object.class))).thenReturn(transactionExecutor);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(recoveryScheduler);
+ when(recoveryScheduler.getExecutor(any(Object.class))).thenReturn(recoveryExecutor);
+ when(managedLedger.getLastConfirmedEntry()).thenReturn(lastConfirmedEntry);
+ when(managedLedger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(managedLedger.newNonDurableCursor(any(), anyString())).thenReturn(managedCursor);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+
+ transactionBuffer = new TopicTransactionBuffer(topic, processor, AbortedTxnProcessor.SnapshotType.Single);
+ }
+
+ private void awaitExecutorsIdle() throws Exception {
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ recoveryExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void close() throws Exception {
+ transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ transactionExecutor.shutdownNow();
+ recoveryExecutor.shutdownNow();
+ assertTrue(transactionExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
index 93d154b8625d..2752d5b672e7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
@@ -18,8 +18,14 @@
*/
package org.apache.pulsar.broker.transaction.buffer.impl;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -28,17 +34,22 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -70,6 +81,83 @@ public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
return new Object[][] { { false }, { true } };
}
+ @Test(dataProvider = "snapshotExists", timeOut = 30_000)
+ public void testRecoveryNotifiesLastPositionQueryOutsideBufferLock(boolean snapshotExists) throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tb-recovery-last-position");
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(topic -> new TopicTransactionBuffer(
+ (PersistentTopic) topic, processor, AbortedTxnProcessor.SnapshotType.Single));
+ try {
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(topicName, true).get(5, TimeUnit.SECONDS).orElseThrow();
+ Position lastConfirmedEntry = topic.getManagedLedger().getLastConfirmedEntry();
+ assertThat(lastConfirmedEntry.getEntryId()).isEqualTo(-1);
+
+ // GetLastMessageId registers this continuation; an empty ledger's lookup can take the topic lock inline.
+ CompletableFuture<Position> lastPositionFuture = topic.checkIfTransactionBufferRecoverCompletely()
+ .thenCompose(__ -> {
+ assertThat(Thread.holdsLock(topic.getTransactionBuffer()))
+ .as("recovery callbacks must run outside the buffer monitor")
+ .isFalse();
+ return topic.getLastDispatchablePosition();
+ });
+
+ recoveryFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+ assertThat(lastPositionFuture.get(5, TimeUnit.SECONDS)).isEqualTo(lastConfirmedEntry);
+ } finally {
+ recoveryFuture.complete(null);
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+
+ @Test(dataProvider = "snapshotExists", timeOut = 30_000)
+ public void testRecoveryCompletesAfterDeletionFails(boolean snapshotExists) throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tb-recovery-delete-failure");
+ TopicName parsedTopicName = TopicName.get(topicName);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CompletableFuture<SchemaVersion> schemaDeleteFuture = new CompletableFuture<>();
+ SchemaRegistryService schemaRegistryService = spy(pulsar.getSchemaRegistryService());
+ doReturn(schemaDeleteFuture).when(schemaRegistryService).deleteSchemaStorage(parsedTopicName.getSchemaName());
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(topic -> new TopicTransactionBuffer(
+ (PersistentTopic) topic, processor, AbortedTxnProcessor.SnapshotType.Single));
+ doReturn(schemaRegistryService).when(pulsar).getSchemaRegistryService();
+ try {
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(topicName, true).get(5, TimeUnit.SECONDS).orElseThrow();
+ ExecutorService transactionExecutor =
+ pulsar.getTransactionExecutorProvider().getExecutor(topic.getTransactionBuffer());
+ // Ensure the recovery continuation is registered before completing its future.
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ CompletableFuture<Void> deleteFuture = topic.delete();
+ verify(schemaRegistryService, timeout(5_000)).deleteSchemaStorage(parsedTopicName.getSchemaName());
+ assertThat(topic.isClosingOrDeleting()).isTrue();
+
+ recoveryFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+ // Let replay run while deletion is pending, regardless of which executor dispatches it.
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ pulsar.getTransactionSnapshotRecoverExecutorProvider().getExecutor(parsedTopicName.getNamespace())
+ .submit(() -> { }).get(5, TimeUnit.SECONDS);
+
+ schemaDeleteFuture.completeExceptionally(new IllegalStateException("schema deletion failed"));
+ assertThatThrownBy(() -> deleteFuture.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(IllegalStateException.class);
+ assertThat(topic.isClosingOrDeleting()).isFalse();
+ topic.getTransactionBuffer().checkIfTBRecoverCompletely().get(5, TimeUnit.SECONDS);
+ } finally {
+ schemaDeleteFuture.completeExceptionally(new IllegalStateException("test finished"));
+ doCallRealMethod().when(pulsar).getSchemaRegistryService();
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+
/**
* While the transaction buffer is recovering, normal publishes don't move the max read position, so they
* don't update the topic's lastMaxReadPositionMovedForwardTimestamp either. When recovery completes, the
(pulsar) 08/14: [fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c7c963cfa6680b8d21044fd949504ec0bd66a256
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 9 05:05:57 2026 +0300
[fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
(cherry picked from commit f3d20b1422fc2dd1d0b3b168b98ec4929aa969dd)
---
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../service/persistent/ReaderMessageTTLTest.java | 99 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 27dcc4a7db85..5eabdaea92ec 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2298,7 +2298,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private void checkMessageExpiryWithoutSharedPosition(int messageTtlInSeconds) {
subscriptions.forEach((__, sub) -> {
- if (!isCompactionSubscription(sub.getName())
+ // TTL must not advance non-durable reader cursors past unread retained messages.
+ if (sub.getCursor().isDurable() && !isCompactionSubscription(sub.getName())
&& (additionalSystemCursorNames.isEmpty()
|| !additionalSystemCursorNames.contains(sub.getName()))) {
sub.expireMessagesAsync(messageTtlInSeconds);
@@ -2343,9 +2344,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// Nothing need to be expired.
return;
}
- // Expire messages by position, which is more efficient.
+ // Expire durable subscriptions by position. Non-durable readers must retain their read position.
subscriptions.forEach((__, sub) -> {
- if (!isCompactionSubscription(sub.getName())
+ if (sub.getCursor().isDurable() && !isCompactionSubscription(sub.getName())
&& (additionalSystemCursorNames.isEmpty()
|| !additionalSystemCursorNames.contains(sub.getName()))) {
// The variable "position" is to mark delete position.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
new file mode 100644
index 000000000000..2e57a883aca0
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.persistent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.awaitility.Awaitility;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ReaderMessageTTLTest extends SharedPulsarBaseTest {
+
+ @DataProvider
+ public Object[][] expiryPaths() {
+ return new Object[][] {{true}, {false}};
+ }
+
+ @Test(dataProvider = "expiryPaths", timeOut = 60000)
+ public void testReaderRetainsUnreadMessagesDuringExpiry(boolean sharedPosition) throws Exception {
+ String topicName = newTopicName();
+ admin.namespaces().setRetention(getNamespace(), new RetentionPolicies(-1, -1));
+ admin.namespaces().setNamespaceMessageTTL(getNamespace(), 1);
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "durable", MessageId.earliest);
+ int messageCount = 20;
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName)
+ .enableBatching(false).create()) {
+ for (int i = 0; i < messageCount; i++) {
+ producer.send("message-" + i);
+ }
+ }
+ try (Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topicName)
+ .subscriptionName("reader").startMessageId(MessageId.earliest).receiverQueueSize(1).create()) {
+ Message<String> first = reader.readNext(10, TimeUnit.SECONDS);
+ assertThat(first).isNotNull();
+ assertThat(first.getValue()).isEqualTo("message-0");
+ PersistentTopic topic = (PersistentTopic) getTopicReference(topicName).orElseThrow();
+ PersistentSubscription readerSubscription = topic.getSubscription("reader");
+ assertThat(readerSubscription.getCursor().isDurable()).isFalse();
+ Position readPosition = readerSubscription.getCursor().getReadPosition();
+ PersistentSubscription durable = topic.getSubscription("durable");
+
+ // Pause the reader until all published messages are old enough to expire.
+ long expiryTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2);
+ Awaitility.await().until(() -> System.currentTimeMillis() > expiryTime);
+ PersistentTopic expiryTopic = topic;
+ if (!sharedPosition) {
+ expiryTopic = spy(topic);
+ // Exercise the fallback used by custom ManagedLedger implementations.
+ doReturn(mock(ManagedLedger.class, delegatesTo(topic.getManagedLedger())))
+ .when(expiryTopic).getManagedLedger();
+ }
+ PersistentTopic topicToCheck = expiryTopic;
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ topicToCheck.checkMessageExpiry();
+ assertThat(durable.getCursor().getNumberOfEntriesInBacklog(false)).isZero();
+ });
+ assertThat(readerSubscription.getCursor().getMarkDeletedPosition())
+ .as("TTL must not acknowledge unread reader messages")
+ .isLessThan(readPosition);
+ for (int i = 1; i < messageCount; i++) {
+ Message<String> message = reader.readNext(10, TimeUnit.SECONDS);
+ assertThat(message).as("retained message %s", i).isNotNull();
+ assertThat(message.getValue()).isEqualTo("message-" + i);
+ }
+ assertThat(reader.hasMessageAvailable()).isFalse();
+ }
+ }
+}
(pulsar) 05/14: [improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b633a2bdc7732854d602eab57015821542de6d89
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 5 04:54:08 2026 +0300
[improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
(cherry picked from commit bc1785dcd867c8e9a03310b84cab5a0a51d14ef5)
---
.../proxy/socket/client/PerformanceClient.java | 23 ++++----
.../socket/client/SimpleTestProducerSocket.java | 13 +++--
.../pulsar/testclient/ManagedLedgerWriter.java | 25 +++++----
.../apache/pulsar/testclient/PerfClientUtils.java | 17 ++++++
.../pulsar/testclient/PerformanceConsumer.java | 38 +++++++-------
.../pulsar/testclient/PerformanceProducer.java | 41 ++++++++-------
.../pulsar/testclient/PerformanceReader.java | 26 +++++----
.../pulsar/testclient/PerformanceTransaction.java | 61 ++++++++++++----------
.../pulsar/testclient/PerfClientUtilsTest.java | 23 ++++++++
9 files changed, 171 insertions(+), 96 deletions(-)
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
index 34370ccf34a7..2636b196ab51 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.proxy.socket.client;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.FileInputStream;
@@ -44,6 +45,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import org.HdrHistogram.Histogram;
import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -68,10 +70,13 @@ import picocli.CommandLine.Parameters;
@Command(name = "websocket-producer", description = "Test pulsar websocket producer performance.")
public class PerformanceClient extends CmdBase {
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
+
+ private final Recorder recorder =
+ new Recorder(SimpleTestProducerSocket.MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
private static IMessageFormatter messageFormatter = null;
@Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
@@ -261,7 +266,7 @@ public class PerformanceClient extends CmdBase {
}
}
- SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket();
+ SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket(recorder);
try {
produceClient.start();
@@ -369,7 +374,7 @@ public class PerformanceClient extends CmdBase {
double rate = messagesSent.sumThenReset() / elapsed;
double throughput = bytesSent.sumThenReset() / elapsed / 1024 / 1024 * 8;
- reportHistogram = SimpleTestProducerSocket.recorder.getIntervalHistogram(reportHistogram);
+ reportHistogram = recorder.getIntervalHistogram(reportHistogram);
log.info(
"Throughput produced: {} msg --- {} msg/s --- {} Mbit/s --- Latency: mean: {} ms - med: {} ms "
@@ -440,7 +445,7 @@ public class PerformanceClient extends CmdBase {
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
@@ -451,8 +456,8 @@ public class PerformanceClient extends CmdBase {
TOTALFORMAT.format(throughput));
}
- private static void printAggregatedStats() {
- Histogram reportHistogram = SimpleTestProducerSocket.recorder.getIntervalHistogram();
+ private void printAggregatedStats() {
+ Histogram reportHistogram = recorder.getIntervalHistogram();
log.info(
"Aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} "
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
index 499cbc951cb0..e08c1f6e79d9 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
@@ -41,14 +41,21 @@ import org.slf4j.LoggerFactory;
@WebSocket
public class SimpleTestProducerSocket {
- public static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
+ // Latencies are recorded in microseconds; values above this ceiling are clamped rather than
+ // rejected, since HdrHistogram throws on an out-of-range value. Package-private so that
+ // PerformanceClient, which owns the shared Recorder, can size it consistently.
+ static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
+
+ // Shared by every socket of one PerformanceClient run so that latencies aggregate.
+ private final Recorder recorder;
private final CountDownLatch closeLatch;
private volatile Session session;
private ConcurrentHashMap<String, Long> startTimeMap = new ConcurrentHashMap<>();
private static final String CONTEXT = "context";
- public SimpleTestProducerSocket() {
+ public SimpleTestProducerSocket(Recorder recorder) {
+ this.recorder = recorder;
this.closeLatch = new CountDownLatch(2);
}
@@ -78,7 +85,7 @@ public class SimpleTestProducerSocket {
startTime = startTimeMap.get(json.get(CONTEXT).getAsString());
}
long latencyNs = endTimeNs - startTime;
- recorder.recordValue(NANOSECONDS.toMicros(latencyNs));
+ recorder.recordValue(Math.min(NANOSECONDS.toMicros(latencyNs), MAX_LATENCY_MICROS));
}
public Session getSession() {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
index a37e254a0ec6..0912d4d6d211 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.util.concurrent.RateLimiter;
@@ -67,13 +68,18 @@ import picocli.CommandLine.Option;
@Command(name = "managed-ledger", description = "Write directly on managed-ledgers")
public class ManagedLedgerWriter extends CmdBase{
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
+ // Latencies are recorded in microseconds. A managed-ledger write slower than this means the
+ // benchmark is broken rather than slow, so values are clamped instead of widening the range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
+
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
@@ -252,7 +258,8 @@ public class ManagedLedgerWriter extends CmdBase{
totalMessagesSent.increment();
totalBytesSent.add(payloadData.length);
- long latencyMicros = NANOSECONDS.toMicros(System.nanoTime() - sendTime);
+ long latencyMicros = Math.min(
+ NANOSECONDS.toMicros(System.nanoTime() - sendTime), MAX_LATENCY_MICROS);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
@@ -392,7 +399,7 @@ public class ManagedLedgerWriter extends CmdBase{
return map;
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
@@ -403,7 +410,7 @@ public class ManagedLedgerWriter extends CmdBase{
TOTALFORMAT.format(throughput));
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.info(
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
index 18d738f6ae3a..2327f007092d 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
@@ -44,6 +44,23 @@ import org.slf4j.Logger;
@UtilityClass
public class PerfClientUtils {
+ /**
+ * Number of significant decimal digits kept by the perf clients' latency histograms.
+ *
+ * <p>HdrHistogram sizes a fixed-range histogram's counts array proportionally to
+ * {@code 2^ceil(log2(2 * 10^digits))}, so every extra digit multiplies the allocation by
+ * roughly 10. At 5 digits (HdrHistogram's maximum) a single {@code Recorder} over the
+ * ranges used here costs 11-22 MB, and since these recorders are static fields the JVM
+ * pays for every perf subcommand, not just the one being run.
+ *
+ * <p>3 digits bounds the error of a reported percentile at 0.1%. That is far below the
+ * run-to-run variance of a benchmark, and finer than the reports resolve anyway: the
+ * microsecond-based tools print milliseconds with {@code %.3f}, and the millisecond-based
+ * ones print whole milliseconds with {@code %d}. It is also HdrHistogram's own recommended
+ * default.
+ */
+ public static final int LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS = 3;
+
private static volatile Consumer<Integer> exitProcedure = System::exit;
public static void setExitProcedure(Consumer<Integer> exitProcedure) {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
index 29d7a2549114..685607df590e 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.util.concurrent.RateLimiter;
@@ -59,28 +60,29 @@ import picocli.CommandLine.Option;
@Command(name = "consume", description = "Test pulsar consumer performance.")
public class PerformanceConsumer extends PerformanceTopicListArguments{
- private static final LongAdder messagesReceived = new LongAdder();
- private static final LongAdder bytesReceived = new LongAdder();
+ private final LongAdder messagesReceived = new LongAdder();
+ private final LongAdder bytesReceived = new LongAdder();
private static final DecimalFormat intFormat = new PaddingDecimalFormat("0", 7);
private static final DecimalFormat dec = new DecimalFormat("0.000");
- private static final LongAdder totalMessagesReceived = new LongAdder();
- private static final LongAdder totalBytesReceived = new LongAdder();
+ private final LongAdder totalMessagesReceived = new LongAdder();
+ private final LongAdder totalBytesReceived = new LongAdder();
- private static final LongAdder totalNumTxnOpenFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenSuccess = new LongAdder();
- private static final LongAdder totalMessageAck = new LongAdder();
- private static final LongAdder totalMessageAckFailed = new LongAdder();
- private static final LongAdder messageAck = new LongAdder();
+ private final LongAdder totalMessageAck = new LongAdder();
+ private final LongAdder totalMessageAckFailed = new LongAdder();
+ private final LongAdder messageAck = new LongAdder();
- private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
- private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
- private static final long MAX_LATENCY = TimeUnit.DAYS.toMillis(10);
- private static final Recorder recorder = new Recorder(MAX_LATENCY, 5);
- private static final Recorder cumulativeRecorder = new Recorder(MAX_LATENCY, 5);
+ private static final long MAX_LATENCY_MILLIS = TimeUnit.DAYS.toMillis(10);
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
+ "one consumer is allowed when subscriptionType is Exclusive",
@@ -276,8 +278,8 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
long latencyMillis = System.currentTimeMillis() - msg.getPublishTime();
if (latencyMillis >= 0) {
- if (latencyMillis >= MAX_LATENCY) {
- latencyMillis = MAX_LATENCY;
+ if (latencyMillis >= MAX_LATENCY_MILLIS) {
+ latencyMillis = MAX_LATENCY_MILLIS;
}
recorder.recordValue(latencyMillis);
cumulativeRecorder.recordValue(latencyMillis);
@@ -569,7 +571,7 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
totalnumMessageAckFailed);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.info(
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
index 5394838f8a51..3a38755a9d85 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -22,6 +22,7 @@ import static java.util.Objects.requireNonNull;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_BATCHING_MAX_MESSAGES;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.collect.Range;
@@ -80,22 +81,28 @@ import picocli.CommandLine.TypeConversionException;
*/
@Command(name = "produce", description = "Test pulsar producer performance.")
public class PerformanceProducer extends PerformanceTopicListArguments{
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder messagesFailed = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder messagesFailed = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
- private static final Recorder recorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ // Publish latencies are recorded in microseconds. A send slower than this means the benchmark is
+ // broken rather than slow, so values are clamped to keep HdrHistogram in range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
- private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
- private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private final Recorder recorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+
+ private final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
private static IMessageFormatter messageFormatter = null;
@@ -650,17 +657,13 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
long now = System.nanoTime();
if (now > warmupEndTime) {
- long latencyMicros = NANOSECONDS.toMicros(now - sendTime);
+ long latencyMicros =
+ Math.min(NANOSECONDS.toMicros(now - sendTime), MAX_LATENCY_MICROS);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
}
}).exceptionally(ex -> {
- // Ignore the exception of recorder since a very large latencyMicros will lead
- // ArrayIndexOutOfBoundsException in AbstractHistogram
Throwable cause = FutureUtil.unwrapCompletionException(ex);
- if (cause instanceof ArrayIndexOutOfBoundsException) {
- return null;
- }
// Ignore the exception when the producer is closed
if (cause instanceof PulsarClientException.AlreadyClosedException) {
return null;
@@ -785,7 +788,7 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
TOTALFORMAT.format(throughput));
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.info(
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
index d4170509a6e5..9eb2fd2c2be4 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.testclient;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import static org.apache.pulsar.testclient.PerfClientUtils.addShutdownHook;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
@@ -49,16 +50,19 @@ import picocli.CommandLine.Option;
@Command(name = "read", description = "Test pulsar reader performance.")
public class PerformanceReader extends PerformanceTopicListArguments {
- private static final LongAdder messagesReceived = new LongAdder();
- private static final LongAdder bytesReceived = new LongAdder();
+ private final LongAdder messagesReceived = new LongAdder();
+ private final LongAdder bytesReceived = new LongAdder();
private static final DecimalFormat intFormat = new PaddingDecimalFormat("0", 7);
private static final DecimalFormat dec = new DecimalFormat("0.000");
- private static final LongAdder totalMessagesReceived = new LongAdder();
- private static final LongAdder totalBytesReceived = new LongAdder();
+ private final LongAdder totalMessagesReceived = new LongAdder();
+ private final LongAdder totalBytesReceived = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
+ private static final long MAX_LATENCY_MILLIS = TimeUnit.DAYS.toMillis(10);
+
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = {"-r", "--rate"}, description = "Simulate a slow message reader (rate in msg/s)")
public double rate = 0;
@@ -126,8 +130,10 @@ public class PerformanceReader extends PerformanceTopicListArguments {
long latencyMillis = System.currentTimeMillis() - msg.getPublishTime();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
+ // Reading a backlog older than the histogram range must not blow up the read loop.
+ long clampedLatencyMillis = Math.min(latencyMillis, MAX_LATENCY_MILLIS);
+ recorder.recordValue(clampedLatencyMillis);
+ cumulativeRecorder.recordValue(clampedLatencyMillis);
}
};
@@ -216,7 +222,7 @@ public class PerformanceReader extends PerformanceTopicListArguments {
PerfClientUtils.closeClient(pulsarClient);
PerfClientUtils.removeAndRunShutdownHook(shutdownHookThread);
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesReceived.sum() / elapsed;
double throughput = totalBytesReceived.sum() / elapsed * 8 / 1024 / 1024;
@@ -227,7 +233,7 @@ public class PerformanceReader extends PerformanceTopicListArguments {
dec.format(throughput));
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.info(
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
index 27f156eb033c..4f6474f0ea34 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import static org.apache.pulsar.testclient.PerfClientUtils.addShutdownHook;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
@@ -68,26 +69,30 @@ import picocli.CommandLine.Option;
@Command(name = "transaction", description = "Test pulsar transaction performance.")
public class PerformanceTransaction extends PerformanceBaseArguments{
- private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
- private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+ private final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
- private static final LongAdder numMessagesAckFailed = new LongAdder();
- private static final LongAdder numMessagesAckSuccess = new LongAdder();
- private static final LongAdder numMessagesSendFailed = new LongAdder();
- private static final LongAdder numMessagesSendSuccess = new LongAdder();
+ private final LongAdder numMessagesAckFailed = new LongAdder();
+ private final LongAdder numMessagesAckSuccess = new LongAdder();
+ private final LongAdder numMessagesSendFailed = new LongAdder();
+ private final LongAdder numMessagesSendSuccess = new LongAdder();
- private static final Recorder messageAckRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder messageAckCumulativeRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ // Send and ack latencies are recorded in microseconds. Anything slower than this means the
+ // benchmark is broken rather than slow, so values are clamped to keep HdrHistogram in range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
- private static final Recorder messageSendRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder messageSendRCumulativeRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private final Recorder messageAckRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder messageAckCumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+
+ private final Recorder messageSendRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder messageSendRCumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
true)
@@ -309,8 +314,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
if (!this.isDisableTransaction) {
consumer.acknowledgeAsync(message.getMessageId(), transaction)
.thenRun(() -> {
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - receiveTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - receiveTime), MAX_LATENCY_MICROS);
messageAckRecorder.recordValue(latencyMicros);
messageAckCumulativeRecorder.recordValue(latencyMicros);
numMessagesAckSuccess.increment();
@@ -327,8 +332,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
});
} else {
consumer.acknowledgeAsync(message).thenRun(() -> {
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - receiveTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - receiveTime), MAX_LATENCY_MICROS);
messageAckRecorder.recordValue(latencyMicros);
messageAckCumulativeRecorder.recordValue(latencyMicros);
numMessagesAckSuccess.increment();
@@ -354,8 +359,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
if (!this.isDisableTransaction) {
producer.newMessage(transaction).value(payloadBytes)
.sendAsync().thenRun(() -> {
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - sendTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - sendTime), MAX_LATENCY_MICROS);
messageSendRecorder.recordValue(latencyMicros);
messageSendRCumulativeRecorder.recordValue(latencyMicros);
numMessagesSendSuccess.increment();
@@ -377,8 +382,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
} else {
producer.newMessage().value(payloadBytes)
.sendAsync().thenRun(() -> {
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - sendTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - sendTime), MAX_LATENCY_MICROS);
messageSendRecorder.recordValue(latencyMicros);
messageSendRCumulativeRecorder.recordValue(latencyMicros);
numMessagesSendSuccess.increment();
@@ -534,7 +539,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
- private static void printTxnAggregatedThroughput(long start) {
+ private void printTxnAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
long numTransactionEndFailed = totalNumEndTxnOpFailed.sum();
long numTransactionEndSuccess = totalNumEndTxnOpSuccess.sum();
@@ -566,7 +571,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
long total = totalNumEndTxnOpFailed.sum() + totalNumEndTxnOpSuccess.sum();
double rate = total / elapsed;
@@ -586,7 +591,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
numMessageSendSuccess);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportAckHistogram = messageAckCumulativeRecorder.getIntervalHistogram();
Histogram reportSendHistogram = messageSendRCumulativeRecorder.getIntervalHistogram();
log.info(
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
index 0c2677db7aa0..2592963f453d 100644
--- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
@@ -18,12 +18,15 @@
*/
package org.apache.pulsar.testclient;
+import static org.assertj.core.api.Assertions.assertThat;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import org.HdrHistogram.Histogram;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.ProxyProtocol;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -166,6 +169,26 @@ public class PerfClientUtilsTest {
Files.deleteIfExists(testConf);
}
}
+
+ /**
+ * The perf clients hold their latency recorders in static fields, so every subcommand allocates them on
+ * startup rather than only the one being run. HdrHistogram grows the counts array by roughly 10x per
+ * significant digit, and at 5 digits these same ranges cost 11-22 MB each. Pin the bound so raising the
+ * precision again fails here instead of silently costing hundreds of megabytes.
+ */
+ @Test
+ public void latencyHistogramsStaySmallAtTheConfiguredPrecision() {
+ long[] rangesUsedByPerfClients = {
+ TimeUnit.HOURS.toMicros(1), // publish / ack / managed-ledger write latency, in microseconds
+ TimeUnit.DAYS.toMillis(10), // end-to-end consume / read latency, in milliseconds
+ };
+ for (long range : rangesUsedByPerfClients) {
+ Histogram histogram = new Histogram(range, PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ assertThat(histogram.getEstimatedFootprintInBytes())
+ .as("histogram footprint for range %d", range)
+ .isLessThan(512 * 1024);
+ }
+ }
}
class PerformanceArgumentsTestDefault extends PerformanceBaseArguments {
(pulsar) 02/14: [improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 77bab3bb4ae829fc7adc7c71b719cd3c419c5175
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Sep 4 14:56:14 2026 -0700
[improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)
(cherry picked from commit 7010231da456da96d49dff50f90225d196451f2a)
---
.../apache/pulsar/common/protocol/Commands.java | 25 ++++++++++++++++++++++
.../pulsar/common/protocol/CommandUtilsTest.java | 24 +++++++++++++++++++++
2 files changed, 49 insertions(+)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index 275b6160061f..1b43686b33a4 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -132,6 +132,13 @@ public class Commands {
public static final short magicCrc32c = 0x0e01;
@SuppressWarnings("checkstyle:ConstantName")
public static final short magicBrokerEntryMetadata = 0x0e02;
+
+ /**
+ * Payloads up to this size are copied into a single buffer when broker entry metadata is prepended; larger ones
+ * are wrapped in a composite instead. Matches the size below which the BookKeeper client copies entries anyway.
+ */
+ @VisibleForTesting
+ static final int BROKER_ENTRY_METADATA_COPY_THRESHOLD = 16 * 1024;
private static final int checksumSize = 4;
@VisibleForTesting
@@ -1761,6 +1768,24 @@ public class Commands {
}
int brokerMetaSize = brokerEntryMetadata.getSerializedSize();
+ int payloadSize = headerAndPayload.readableBytes();
+
+ if (payloadSize <= BROKER_ENTRY_METADATA_COPY_THRESHOLD) {
+ // For small payloads, produce a single contiguous buffer. A composite reaching the socket goes through
+ // nioBuffers(), which allocates two NIO views per component on every flush (BK write and each consumer
+ // dispatch), and makes every metadata parse walk the components. At this size the copy is cheaper than
+ // that, and it happens once at publish while the entry is written and parsed many more times.
+ int totalSize = 6 + brokerMetaSize + payloadSize;
+ ByteBuf entry = PulsarByteBufAllocator.DEFAULT.buffer(totalSize, totalSize);
+ entry.writeShort(Commands.magicBrokerEntryMetadata);
+ entry.writeInt(brokerMetaSize);
+ brokerEntryMetadata.writeTo(entry);
+ entry.writeBytes(headerAndPayload, headerAndPayload.readerIndex(), payloadSize);
+ headerAndPayload.release();
+ return entry;
+ }
+
+ // For large payloads the copy would cost more than the per-flush view allocations it saves.
ByteBuf brokerMeta =
PulsarByteBufAllocator.DEFAULT.buffer(brokerMetaSize + 6, brokerMetaSize + 6);
brokerMeta.writeShort(Commands.magicBrokerEntryMetadata);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
index baea2b3f0279..8bf2d6f389ca 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
@@ -148,6 +148,30 @@ public class CommandUtilsTest {
dataWithBrokerEntryMetadata.release();
}
+ @Test
+ public void testAddBrokerEntryMetadataLargePayload() throws Exception {
+ int mockBatchSize = 10;
+ byte[] data = new byte[Commands.BROKER_ENTRY_METADATA_COPY_THRESHOLD + 1];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) i;
+ }
+ ByteBuf byteBuf = PulsarByteBufAllocator.DEFAULT.buffer(data.length, data.length);
+ byteBuf.writeBytes(data);
+ ByteBuf dataWithBrokerEntryMetadata =
+ Commands.addBrokerEntryMetadata(byteBuf, getBrokerEntryMetadataInterceptors(), mockBatchSize);
+ assertTrue(dataWithBrokerEntryMetadata instanceof CompositeByteBuf);
+
+ BrokerEntryMetadata brokerMetadata =
+ Commands.parseBrokerEntryMetadataIfExist(dataWithBrokerEntryMetadata);
+ assertEquals(brokerMetadata.getIndex(), mockBatchSize - 1);
+ assertEquals(data.length, dataWithBrokerEntryMetadata.readableBytes());
+
+ byte[] content = new byte[dataWithBrokerEntryMetadata.readableBytes()];
+ dataWithBrokerEntryMetadata.readBytes(content);
+ assertEquals(content, data);
+ dataWithBrokerEntryMetadata.release();
+ }
+
@Test
public void testSkipBrokerEntryMetadata() throws Exception {
String data = "test-message";
(pulsar) 10/14: [fix][client] Release the reserved memory only once when failing a send in a terminal state (#26476)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6a5b19d640f10037c71be5bc75ac6c7a55c540ce
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Wed Sep 9 10:35:36 2026 +0800
[fix][client] Release the reserved memory only once when failing a send in a terminal state (#26476)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit 698192dca86b28af23e623a76acc4dfa1b361f9b)
---
.../apache/pulsar/client/impl/ProducerImpl.java | 2 +-
.../pulsar/client/impl/ProducerImplTest.java | 46 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 1 deletion(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index bee2e9b4a7f7..130dfb64fea9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -2507,8 +2507,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (state == State.Terminated || state == State.Closed || state == State.ProducerFenced) {
// The producer is in a terminal state and will never reconnect. Fail the message immediately
// rather than leaving it stuck in pendingMessages until sendTimeout.
+ // releaseSemaphoreForSendOp() also gives the reserved memory back, so it must not be released again.
releaseSemaphoreForSendOp(op);
- client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
op.sendComplete(getTerminalException(state));
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
index 0fe177b775cb..0b832e414d9d 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
@@ -26,6 +26,7 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import java.nio.ByteBuffer;
+import java.util.Collections;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
@@ -144,4 +145,49 @@ public class ProducerImplTest {
assertEquals(pendingQueue.peek().sequenceId, 2L,
"Retry Op SequenceId should match with the one in pendingQueue");
}
+
+ /**
+ * When the producer is in a terminal state, {@link ProducerImpl#processOpSendMsg} fails the message right away.
+ * The memory reserved for that message must be given back to the {@link MemoryLimitController} exactly once:
+ * releasing it twice makes {@code currentUsage} drift below zero and permanently disables the client memory
+ * limit.
+ */
+ @Test
+ public void testProcessOpSendMsgInTerminalStateReleasesMemoryOnce() throws Exception {
+ for (ProducerImpl.State state : new ProducerImpl.State[] {
+ ProducerImpl.State.Terminated, ProducerImpl.State.Closed, ProducerImpl.State.ProducerFenced}) {
+ @SuppressWarnings("unchecked")
+ ProducerImpl<byte[]> producer = Mockito.mock(ProducerImpl.class, Mockito.CALLS_REAL_METHODS);
+ // Disable batching, so that releaseSemaphoreForSendOp() releases a single permit
+ Mockito.doReturn(false).when(producer).isBatchMessagingEnabled();
+ // The semaphore is not under test
+ Mockito.doNothing().when(producer).semaphoreRelease(Mockito.anyInt());
+
+ MemoryLimitController memoryLimitController = new MemoryLimitController(1024 * 1024);
+ PulsarClientImpl client = Mockito.mock(PulsarClientImpl.class);
+ Mockito.when(client.getMemoryLimitController()).thenReturn(memoryLimitController);
+ FieldUtils.writeField(producer, "client", client, true);
+
+ int uncompressedSize = 128;
+ MessageImpl<?> msg = Mockito.mock(MessageImpl.class);
+ Mockito.when(msg.getUncompressedSize()).thenReturn(uncompressedSize);
+ // Build the op through the batch factory so that op.msg is null and the message size check is skipped
+ ProducerImpl.OpSendMsg op = ProducerImpl.OpSendMsg.create(
+ Mockito.mock(LatencyHistogram.class),
+ Collections.<MessageImpl<?>>singletonList(msg),
+ Mockito.mock(ByteBufPair.class),
+ 1L,
+ Mockito.mock(SendCallback.class),
+ 0);
+
+ memoryLimitController.forceReserveMemory(op.uncompressedSize);
+ assertEquals(memoryLimitController.currentUsage(), uncompressedSize);
+
+ producer.setState(state);
+ producer.processOpSendMsg(op);
+
+ assertEquals(memoryLimitController.currentUsage(), 0,
+ "The memory reserved for the message must be released exactly once in state " + state);
+ }
+ }
}
(pulsar) 12/14: [improve][cli] Default the pulsar-perf client memory limit to half of the JVM max direct memory (#26491)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 59311c967c40b06ce4daee28ab0690e288d7bf35
Author: Alvaro <10...@users.noreply.github.com>
AuthorDate: Wed Sep 9 08:45:46 2026 +0200
[improve][cli] Default the pulsar-perf client memory limit to half of the JVM max direct memory (#26491)
(cherry picked from commit 0b56d9d33a1b66d025bdfbc64ebd85e5a35b6ea7)
---
.../testclient/PerformanceBaseArguments.java | 17 +++++++++--
.../testclient/PerformanceBaseArgumentsTest.java | 34 ++++++++++++++++++++++
2 files changed, 49 insertions(+), 2 deletions(-)
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java
index ee79066c32f9..7b871dfd8927 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter;
import org.apache.pulsar.client.api.ProxyProtocol;
+import org.apache.pulsar.common.util.DirectMemoryUtils;
import picocli.CommandLine.Option;
/**
@@ -106,9 +107,21 @@ public abstract class PerformanceBaseArguments extends CmdBase{
@Option(names = { "--auth_plugin" }, description = "Authentication plugin class name", hidden = true)
public String deprecatedAuthPluginClassName;
+ /**
+ * Default Pulsar client memory limit for the performance tools: half of the JVM's max direct memory.
+ *
+ * <p>The performance tools do not use direct memory for anything other than Netty, so bounding the client
+ * at a fraction of what the JVM can actually allocate keeps the tool from dying with
+ * {@code OutOfDirectMemoryError} while still letting it use the memory it was given. Being proportional
+ * rather than a fixed value also avoids silently changing results for runs that were sized with a larger
+ * {@code -XX:MaxDirectMemorySize}.
+ */
+ public static final long DEFAULT_MEMORY_LIMIT_BYTES = (long) (0.5d * DirectMemoryUtils.jvmMaxDirectMemory());
+
@Option(names = { "-ml", "--memory-limit", }, description = "Configure the Pulsar client memory limit "
- + "(eg: 32M, 64M)", converter = ByteUnitToLongConverter.class)
- public long memoryLimit;
+ + "(eg: 32M, 64M). Defaults to half of the JVM's max direct memory. Use 0 to disable the limit.",
+ converter = ByteUnitToLongConverter.class)
+ public long memoryLimit = DEFAULT_MEMORY_LIMIT_BYTES;
public PerformanceBaseArguments(String cmdName) {
super(cmdName);
}
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java
index 78cee5c22154..50e57b0428fc 100644
--- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pulsar.common.util.DirectMemoryUtils;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -249,6 +250,39 @@ public class PerformanceBaseArgumentsTest {
// Act
baseArgument.parseCLI();
+ // Assert
+ assertEquals(baseArgument.memoryLimit, PerformanceBaseArguments.DEFAULT_MEMORY_LIMIT_BYTES);
+ assertEquals(baseArgument.memoryLimit, (long) (0.5d * DirectMemoryUtils.jvmMaxDirectMemory()));
+ }
+ }
+
+ @Test
+ public void testMemoryLimitCanBeDisabled() throws Exception {
+ for (String cmd : List.of(
+ "pulsar-perf read",
+ "pulsar-perf produce",
+ "pulsar-perf consume",
+ "pulsar-perf transaction"
+ )) {
+ // Arrange
+ final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") {
+ @Override
+ public void run() throws Exception {
+
+ }
+
+ };
+ String confFile = "./src/test/resources/perf_client1.conf";
+ Properties prop = new Properties(System.getProperties());
+ try (FileInputStream fis = new FileInputStream(confFile)) {
+ prop.load(fis);
+ }
+ baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ baseArgument.parse(new String[]{"-ml", "0"});
+
+ // Act
+ baseArgument.parseCLI();
+
// Assert
assertEquals(baseArgument.memoryLimit, 0L);
}
(pulsar) 07/14: [fix][broker] Fix NPE in ManagedLedgerInterceptorImpl when AppendIndexMetadataInterceptor isn't configured (#26497)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e6087f0a9ba4e8caae0db2ad975f917b79d52fe1
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Sep 8 17:56:02 2026 +0300
[fix][broker] Fix NPE in ManagedLedgerInterceptorImpl when AppendIndexMetadataInterceptor isn't configured (#26497)
(cherry picked from commit bef7b8f912ca72f56c0c74d69a6b6ea60d226996)
---
.../intercept/ManagedLedgerInterceptorImpl.java | 8 +-
.../ManagedLedgerInterceptorImplTest.java | 111 +++++++++++++++++++++
2 files changed, 118 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
index a86b49d627ff..f2cff8640115 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
@@ -30,6 +30,7 @@ import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
import org.apache.pulsar.common.protocol.Commands;
+import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,7 +39,7 @@ public class ManagedLedgerInterceptorImpl implements ManagedLedgerInterceptor {
private static final String INDEX = "index";
private final Set<BrokerEntryMetadataInterceptor> brokerEntryMetadataInterceptors;
- private final AppendIndexMetadataInterceptor appendIndexMetadataInterceptor;
+ private final @Nullable AppendIndexMetadataInterceptor appendIndexMetadataInterceptor;
private final Set<ManagedLedgerPayloadProcessor.Processor> inputProcessors;
private final Set<ManagedLedgerPayloadProcessor.Processor> outputProcessors;
@@ -112,6 +113,11 @@ public class ManagedLedgerInterceptorImpl implements ManagedLedgerInterceptor {
@Override
public CompletableFuture<Void> onManagedLedgerLastLedgerInitialize(String name, LastEntryHandle lh) {
+ if (appendIndexMetadataInterceptor == null) {
+ // there's no index generator to recover when the AppendIndexMetadataInterceptor isn't configured,
+ // so reading the last entry would be pointless
+ return CompletableFuture.completedFuture(null);
+ }
return lh.readLastEntryAsync().thenAccept(lastEntryOptional -> {
if (lastEntryOptional.isPresent()) {
Entry lastEntry = lastEntryOptional.get();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
index f2e956ac0067..e46beff0df06 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.intercept;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
@@ -28,9 +29,12 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import lombok.Cleanup;
@@ -41,12 +45,14 @@ import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
+import org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
@@ -55,6 +61,7 @@ import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker")
@@ -248,6 +255,110 @@ public class ManagedLedgerInterceptorImplTest extends MockedBookKeeperTestCase
ledger.close();
}
+ @DataProvider(name = "interceptorSetsWithoutIndexInterceptor")
+ public Object[][] interceptorSetsWithoutIndexInterceptor() {
+ return new Object[][]{
+ // brokerEntryMetadataInterceptors holds only the timestamp interceptor
+ {Set.of(new AppendBrokerTimestampMetadataInterceptor())},
+ // brokerEntryMetadataInterceptors is empty: the interceptor is installed because
+ // brokerEntryPayloadProcessors is configured
+ {Collections.<BrokerEntryMetadataInterceptor>emptySet()}
+ };
+ }
+
+ /**
+ * When no AppendIndexMetadataInterceptor is configured there is no index generator to recover into, so
+ * reading the last entry must be skipped altogether. Reading it used to dereference the null
+ * appendIndexMetadataInterceptor whenever the last entry still carried a BrokerEntryMetadata index,
+ * which failed the managed ledger initialization with a ManagedLedgerInterceptException.
+ */
+ @Test(dataProvider = "interceptorSetsWithoutIndexInterceptor", timeOut = 30000)
+ public void testLastLedgerInitializeSkipsIndexRecoveryWithoutIndexInterceptor(
+ Set<BrokerEntryMetadataInterceptor> interceptors) throws Exception {
+ ManagedLedgerInterceptorImpl interceptor = new ManagedLedgerInterceptorImpl(interceptors, null);
+ ByteBuf lastEntryBuffer = lastEntryBufferWithIndex();
+ try {
+ AtomicBoolean lastEntryRead = new AtomicBoolean();
+ interceptor.onManagedLedgerLastLedgerInitialize("my_test_ledger", () -> {
+ lastEntryRead.set(true);
+ return CompletableFuture.completedFuture(
+ Optional.of(EntryImpl.create(1L, 0L, lastEntryBuffer)));
+ }).get(5, TimeUnit.SECONDS);
+
+ assertThat(lastEntryRead).as("last entry was read").isFalse();
+ assertThat(interceptor.getIndex()).as("recovered index").isEqualTo(-1L);
+ } finally {
+ lastEntryBuffer.release();
+ }
+ }
+
+ @Test(timeOut = 30000)
+ public void testLastLedgerInitializeRecoversIndexWithIndexInterceptor() throws Exception {
+ ManagedLedgerInterceptorImpl interceptor =
+ new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors(), null);
+ ByteBuf lastEntryBuffer = lastEntryBufferWithIndex();
+ try {
+ AtomicBoolean lastEntryRead = new AtomicBoolean();
+ interceptor.onManagedLedgerLastLedgerInitialize("my_test_ledger", () -> {
+ lastEntryRead.set(true);
+ return CompletableFuture.completedFuture(
+ Optional.of(EntryImpl.create(1L, 0L, lastEntryBuffer)));
+ }).get(5, TimeUnit.SECONDS);
+
+ assertThat(lastEntryRead).as("last entry was read").isTrue();
+ assertThat(interceptor.getIndex()).as("recovered index").isEqualTo(99L);
+ } finally {
+ lastEntryBuffer.release();
+ }
+ }
+
+ /**
+ * Reopens a managed ledger whose entries were written while AppendIndexMetadataInterceptor was
+ * configured, with a configuration where it no longer is. This is how the null
+ * appendIndexMetadataInterceptor is reached in production.
+ */
+ @Test(timeOut = 30000)
+ public void testRecoveryIndexAfterIndexInterceptorRemovedFromConfiguration() throws Exception {
+ final int mockBatchSize = 2;
+ final String ledgerName = "my_recovery_index_without_index_interceptor";
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setManagedLedgerInterceptor(
+ new ManagedLedgerInterceptorImpl(getBrokerEntryMetadataInterceptors(), null));
+ ManagedLedger ledger = factory.open(ledgerName, config);
+ ledger.addEntry("dummy-entry-1".getBytes(StandardCharsets.UTF_8), mockBatchSize);
+ assertThat(((ManagedLedgerInterceptorImpl) ledger.getManagedLedgerInterceptor()).getIndex())
+ .as("index of the entry written with the index interceptor configured")
+ .isEqualTo(mockBatchSize - 1);
+ ledger.close();
+
+ // reopen with AppendIndexMetadataInterceptor removed from the configuration while the last entry of
+ // the ledger still carries a BrokerEntryMetadata index
+ ManagedLedgerInterceptorImpl interceptorWithoutIndex = new ManagedLedgerInterceptorImpl(
+ Set.of(new AppendBrokerTimestampMetadataInterceptor()), null);
+ ManagedLedgerConfig configWithoutIndex = new ManagedLedgerConfig();
+ configWithoutIndex.setManagedLedgerInterceptor(interceptorWithoutIndex);
+
+ @Cleanup("shutdown")
+ ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedger reopenedLedger = factory2.open(ledgerName, configWithoutIndex);
+ assertThat(reopenedLedger.getNumberOfEntries()).as("entries in the reopened ledger").isEqualTo(1L);
+ assertThat(interceptorWithoutIndex.getIndex()).as("recovered index").isEqualTo(-1L);
+ reopenedLedger.close();
+ }
+
+ // builds an entry buffer carrying a BrokerEntryMetadata index, as a broker configured with
+ // AppendIndexMetadataInterceptor would have written it (100 messages in the entry means index 99)
+ private static ByteBuf lastEntryBufferWithIndex() {
+ ByteBuf entryBuffer = Commands.addBrokerEntryMetadata(
+ Unpooled.wrappedBuffer("message".getBytes(StandardCharsets.UTF_8)),
+ getBrokerEntryMetadataInterceptors(), 100);
+ BrokerEntryMetadata metadata = Commands.peekBrokerEntryMetadataIfExist(entryBuffer);
+ assertThat(metadata).as("broker entry metadata of the entry under test").isNotNull();
+ assertThat(metadata.getIndex()).as("index written into the broker entry metadata").isEqualTo(99L);
+ return entryBuffer;
+ }
+
@Test
public void testFindPositionByIndex() throws Exception {
final int mockBatchSize = 2;
(pulsar) 01/14: [improve][broker] Avoid iterator allocation in BrokerInterceptors hot paths and restore configured ordering (#26462)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d144107e5d658f91a8be252fcdd1d1d9aa4eb715
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Sep 4 14:30:15 2026 -0700
[improve][broker] Avoid iterator allocation in BrokerInterceptors hot paths and restore configured ordering (#26462)
(cherry picked from commit 46182ff0bbffe2d3fcf320a42d73a3b4c81ffb6a)
---
.../broker/intercept/BrokerInterceptors.java | 55 ++++++++++++----------
.../org/apache/pulsar/broker/web/WebService.java | 2 +-
.../broker/intercept/BrokerInterceptorTest.java | 2 +-
.../intercept/ExceptionsBrokerInterceptorTest.java | 2 +-
4 files changed, 33 insertions(+), 28 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
index 3c448529de9e..8bd1b6508300 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java
@@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import java.io.IOException;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
@@ -46,10 +47,12 @@ import org.apache.pulsar.common.intercept.InterceptException;
@Slf4j
public class BrokerInterceptors implements BrokerInterceptor {
- private final Map<String, BrokerInterceptorWithClassLoader> interceptors;
+ // Interceptors in configured order. An array is used because iterating it does not allocate an iterator,
+ // unlike Map.values(), which matters on the per-message hot paths (onMessagePublish, beforeSendMessage, ...).
+ private final BrokerInterceptorWithClassLoader[] interceptors;
public BrokerInterceptors(Map<String, BrokerInterceptorWithClassLoader> interceptors) {
- this.interceptors = interceptors;
+ this.interceptors = interceptors.values().toArray(new BrokerInterceptorWithClassLoader[0]);
}
/**
@@ -91,7 +94,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
});
if (!orderedInterceptorMap.isEmpty()) {
- return new BrokerInterceptors(Map.copyOf(orderedInterceptorMap));
+ return new BrokerInterceptors(orderedInterceptorMap);
} else {
return null;
}
@@ -102,7 +105,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
ByteBuf headersAndPayload,
Topic.PublishContext publishContext) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onMessagePublish(producer, headersAndPayload, publishContext);
}
}
@@ -114,7 +117,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
long[] ackSet,
MessageMetadata msgMetadata) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.beforeSendMessage(subscription, entry, ackSet, msgMetadata);
}
}
@@ -127,7 +130,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
MessageMetadata msgMetadata,
Consumer consumer) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.beforeSendMessage(subscription, entry, ackSet, msgMetadata, consumer);
}
}
@@ -138,7 +141,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
Consumer consumer,
Map<String, String> metadata) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.consumerCreated(
cnx,
consumer,
@@ -152,7 +155,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
Consumer consumer,
Map<String, String> metadata) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.consumerClosed(cnx, consumer, metadata);
}
}
@@ -162,7 +165,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
public void producerCreated(ServerCnx cnx, Producer producer,
Map<String, String> metadata){
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.producerCreated(cnx, producer, metadata);
}
}
@@ -173,7 +176,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
Producer producer,
Map<String, String> metadata) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.producerClosed(cnx, producer, metadata);
}
}
@@ -183,7 +186,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId,
long entryId, Topic.PublishContext publishContext) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.messageProduced(cnx, producer, startTimeNs, ledgerId, entryId, publishContext);
}
}
@@ -193,7 +196,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
public void messageDispatched(ServerCnx cnx, Consumer consumer, long ledgerId,
long entryId, ByteBuf headersAndPayload) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.messageDispatched(cnx, consumer, ledgerId, entryId, headersAndPayload);
}
}
@@ -203,7 +206,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
public void messageAcked(ServerCnx cnx, Consumer consumer,
CommandAck ackCmd) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.messageAcked(cnx, consumer, ackCmd);
}
}
@@ -212,7 +215,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
@Override
public void txnOpened(long tcId, String txnID) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.txnOpened(tcId, txnID);
}
}
@@ -221,7 +224,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
@Override
public void txnEnded(String txnID, long txnAction) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.txnEnded(txnID, txnAction);
}
}
@@ -231,7 +234,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
@Override
public void onConnectionCreated(ServerCnx cnx) {
if (interceptorsEnabled()) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onConnectionCreated(cnx);
}
}
@@ -239,21 +242,21 @@ public class BrokerInterceptors implements BrokerInterceptor {
@Override
public void onPulsarCommand(BaseCommand command, ServerCnx cnx) throws InterceptException {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onPulsarCommand(command, cnx);
}
}
@Override
public void onConnectionClosed(ServerCnx cnx) {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onConnectionClosed(cnx);
}
}
@Override
public void onWebserviceRequest(ServletRequest request) throws IOException, ServletException, InterceptException {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onWebserviceRequest(request);
}
}
@@ -261,29 +264,31 @@ public class BrokerInterceptors implements BrokerInterceptor {
@Override
public void onWebserviceResponse(ServletRequest request, ServletResponse response)
throws IOException, ServletException {
- for (BrokerInterceptorWithClassLoader value : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
value.onWebserviceResponse(request, response);
}
}
@Override
public void initialize(PulsarService pulsarService) throws Exception {
- for (BrokerInterceptorWithClassLoader v : interceptors.values()) {
+ for (BrokerInterceptorWithClassLoader v : interceptors) {
v.initialize(pulsarService);
}
}
@Override
public void close() {
- interceptors.values().forEach(BrokerInterceptorWithClassLoader::close);
+ for (BrokerInterceptorWithClassLoader value : interceptors) {
+ value.close();
+ }
}
private boolean interceptorsEnabled() {
- return interceptors != null && !interceptors.isEmpty();
+ return interceptors.length > 0;
}
@VisibleForTesting
- public Map<String, BrokerInterceptorWithClassLoader> getInterceptors() {
- return interceptors;
+ public List<BrokerInterceptorWithClassLoader> getInterceptors() {
+ return List.of(interceptors);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index 9ef193ff9f13..3c830e01ab38 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -270,7 +270,7 @@ public class WebService implements AutoCloseable {
// hold all interceptors, instead we need to create a `ProcessHandlerFilter` for each `interceptor`.
if (pulsarService.getBrokerInterceptor() instanceof BrokerInterceptors) {
for (BrokerInterceptor interceptor: ((BrokerInterceptors) pulsarService.getBrokerInterceptor())
- .getInterceptors().values()) {
+ .getInterceptors()) {
filterHolders.add(new FilterHolder(new ProcessHandlerFilter(interceptor)));
}
} else {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
index 72f1c4f50094..d1fd46068c19 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java
@@ -104,7 +104,7 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
private CounterBrokerInterceptor getCounterBrokerInterceptor() {
BrokerInterceptor brokerInterceptor = pulsar.getBrokerInterceptor();
BrokerInterceptorWithClassLoader brokerInterceptorWithClassLoader =
- ((BrokerInterceptors) brokerInterceptor).getInterceptors().get(CounterBrokerInterceptor.NAME);
+ ((BrokerInterceptors) brokerInterceptor).getInterceptors().get(0);
return (CounterBrokerInterceptor) brokerInterceptorWithClassLoader.getInterceptor();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ExceptionsBrokerInterceptorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ExceptionsBrokerInterceptorTest.java
index 82078da7612f..bc18b9ad6794 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ExceptionsBrokerInterceptorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ExceptionsBrokerInterceptorTest.java
@@ -77,7 +77,7 @@ public class ExceptionsBrokerInterceptorTest extends ProducerConsumerBase {
BrokerInterceptors listener = (BrokerInterceptors) pulsar.getBrokerInterceptor();
assertNotNull(listener);
- BrokerInterceptorWithClassLoader brokerInterceptor = listener.getInterceptors().get(interceptorName);
+ BrokerInterceptorWithClassLoader brokerInterceptor = listener.getInterceptors().get(0);
assertNotNull(brokerInterceptor);
BrokerInterceptor interceptor = brokerInterceptor.getInterceptor();
assertTrue(interceptor instanceof ExceptionsBrokerInterceptor);
(pulsar) 09/14: [fix][admin] Reject empty cluster migration URL: fix inverted ClusterUrl.isEmpty() (#26473)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 62721fefaf28801362a3e50cf575cd3c1d8ff849
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Wed Sep 9 10:34:18 2026 +0800
[fix][admin] Reject empty cluster migration URL: fix inverted ClusterUrl.isEmpty() (#26473)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit 98b08cc58785533ccaab2f43c55a369c096e809c)
---
.../org/apache/pulsar/broker/admin/AdminTest.java | 18 ++++++-
.../common/policies/data/ClusterPolicies.java | 11 +++-
.../common/policies/data/ClusterUrlTest.java | 60 ++++++++++++++++++++++
3 files changed, 86 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
index 88759522e6cb..6c2e20e5e194 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java
@@ -80,6 +80,8 @@ import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ClusterDataImpl;
+import org.apache.pulsar.common.policies.data.ClusterPolicies;
+import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
import org.apache.pulsar.common.policies.data.ErrorData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationDataImpl;
import org.apache.pulsar.common.policies.data.Policies;
@@ -289,6 +291,20 @@ public class AdminTest extends MockedPulsarServiceBaseTest {
assertEquals(asyncRequests(ctx -> clusters.getCluster(ctx, "use")),
ClusterData.builder().serviceUrl("http://new-broker.messaging.use.example.com:8080").build());
+ // Marking a cluster as migrated without any target url must be rejected
+ try {
+ asyncRequests(ctx -> clusters.updateClusterMigration(ctx, "use", true, new ClusterUrl()));
+ fail("should have failed");
+ } catch (RestException e) {
+ assertEquals(e.getResponse().getStatus(), Status.BAD_REQUEST.getStatusCode());
+ }
+ // ... while a target url of any kind is accepted
+ ClusterUrl migratedUrl = new ClusterUrl(null, null, "pulsar://green.example.com:6650", null);
+ asyncRequests(ctx -> clusters.updateClusterMigration(ctx, "use", true, migratedUrl));
+ ClusterPolicies migration = (ClusterPolicies) asyncRequests(ctx -> clusters.getClusterMigration(ctx, "use"));
+ assertTrue(migration.isMigrated());
+ assertEquals(migration.getMigratedClusterUrl(), migratedUrl);
+
try {
asyncRequests(ctx -> clusters.getNamespaceIsolationPolicies(ctx, "use"));
fail("should have failed");
@@ -456,7 +472,7 @@ public class AdminTest extends MockedPulsarServiceBaseTest {
} catch (RestException e) {
assertEquals(e.getResponse().getStatus(), Status.PRECONDITION_FAILED.getStatusCode());
}
- verify(clusters, times(24)).validateSuperUserAccessAsync();
+ verify(clusters, times(26)).validateSuperUserAccessAsync();
}
@Test
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ClusterPolicies.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ClusterPolicies.java
index b95f6bb19ce2..aad181dafcc9 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ClusterPolicies.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ClusterPolicies.java
@@ -53,9 +53,16 @@ public interface ClusterPolicies {
String brokerServiceUrl;
String brokerServiceUrlTls;
+ /**
+ * @return true when no URL of any kind (HTTP or binary, plain or TLS) is configured.
+ */
public boolean isEmpty() {
- return serviceUrl != null && serviceUrlTls != null && brokerServiceUrl == null
- && brokerServiceUrlTls == null;
+ return isBlank(serviceUrl) && isBlank(serviceUrlTls)
+ && isBlank(brokerServiceUrl) && isBlank(brokerServiceUrlTls);
+ }
+
+ private static boolean isBlank(String s) {
+ return s == null || s.isBlank();
}
}
}
diff --git a/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ClusterUrlTest.java b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ClusterUrlTest.java
new file mode 100644
index 000000000000..672bc3051247
--- /dev/null
+++ b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ClusterUrlTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.common.policies.data;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
+import org.testng.annotations.Test;
+
+public class ClusterUrlTest {
+
+ @Test
+ public void testNoUrlIsEmpty() {
+ // The broker rejects "migrated=true" when the target ClusterUrl is empty. A ClusterUrl with
+ // no URL at all is the canonical empty value and must be reported as such.
+ assertTrue(new ClusterUrl().isEmpty());
+ assertTrue(new ClusterUrl(null, null, null, null).isEmpty());
+ }
+
+ @Test
+ public void testBlankUrlsAreEmpty() {
+ assertTrue(new ClusterUrl("", "", "", "").isEmpty());
+ assertTrue(new ClusterUrl(" ", null, "", null).isEmpty());
+ }
+
+ @Test
+ public void testHttpOnlyUrlIsNotEmpty() {
+ // Only the HTTP endpoints configured: still a URL, so not empty.
+ assertFalse(new ClusterUrl("http://green:8080", "https://green:8443", null, null).isEmpty());
+ assertFalse(new ClusterUrl("http://green:8080", null, null, null).isEmpty());
+ }
+
+ @Test
+ public void testBrokerOnlyUrlIsNotEmpty() {
+ assertFalse(new ClusterUrl(null, null, "pulsar://green:6650", null).isEmpty());
+ assertFalse(new ClusterUrl(null, null, null, "pulsar+ssl://green:6651").isEmpty());
+ }
+
+ @Test
+ public void testFullUrlIsNotEmpty() {
+ assertFalse(new ClusterUrl("http://green:8080", "https://green:8443",
+ "pulsar://green:6650", "pulsar+ssl://green:6651").isEmpty());
+ }
+}