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 16:32:13 UTC
(pulsar) branch branch-4.0 updated (ef07d56dc791 -> fb093d27ec95)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from ef07d56dc791 [fix][sec] Bump google.golang.org/grpc from 1.82.1 to 1.83.1 in /pulsar-function-go (#26446)
new 398a73dd206d [improve][broker] Avoid iterator allocation in BrokerInterceptors hot paths and restore configured ordering (#26462)
new 57375e1e3213 [improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)
new ea35cfe1037b [fix][cli] pulsar-perf: register subcommand classes with picocli instead of instances (#26467)
new 70a6914befb4 [fix][cli] Apply PULSAR_MEM and PULSAR_GC in bin/pulsar-perf (#26465)
new 100ce94050c4 [improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
new a5e9037a433a [fix][broker] Fix NPE in ManagedLedgerInterceptorImpl when AppendIndexMetadataInterceptor isn't configured (#26497)
new 4505bf6484a1 [fix][broker] Prevent automatic TTL expiry from skipping reader messages (#26505)
new 57b493c51258 [fix][admin] Reject empty cluster migration URL: fix inverted ClusterUrl.isEmpty() (#26473)
new e3842dee2bc8 [fix][client] Release the reserved memory only once when failing a send in a terminal state (#26476)
new fd38dba8fd43 [improve][cli] Default the pulsar-perf client memory limit to half of the JVM max direct memory (#26491)
new fb093d27ec95 [fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
The 11 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 ++
.../broker/intercept/BrokerInterceptors.java | 65 +++--
.../intercept/ManagedLedgerInterceptorImpl.java | 8 +-
.../broker/service/persistent/PersistentTopic.java | 7 +-
.../impl/AbstractSnapshotAbortedTxnProcessor.java | 186 +++++++++++++
.../SingleSnapshotAbortedTxnProcessorImpl.java | 46 +---
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 105 +++----
.../buffer/impl/TopicTransactionBuffer.java | 212 +++++++++++----
.../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 | 117 ++++++++
.../pulsar/broker/transaction/TransactionTest.java | 7 +-
.../AbstractSnapshotAbortedTxnProcessorTest.java | 302 +++++++++++++++++++++
...napshotSegmentAbortedTxnProcessorCloseTest.java | 173 ++++++++++++
.../impl/TopicTransactionBufferCloseTest.java | 283 +++++++++++++++++++
.../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 +++
38 files changed, 1979 insertions(+), 311 deletions(-)
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) 10/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fd38dba8fd43cb43a0acdff2800d7e1803159be9
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) 01/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 398a73dd206d8e291d80316f3ce14c35f6152e87
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 | 65 ++++++++++++----------
.../org/apache/pulsar/broker/web/WebService.java | 2 +-
.../broker/intercept/BrokerInterceptorTest.java | 2 +-
.../intercept/ExceptionsBrokerInterceptorTest.java | 2 +-
4 files changed, 38 insertions(+), 33 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 30d1874a9729..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
@@ -19,9 +19,10 @@
package org.apache.pulsar.broker.intercept;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableMap;
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]);
}
/**
@@ -66,7 +69,8 @@ public class BrokerInterceptors implements BrokerInterceptor {
BrokerInterceptorUtils.searchForInterceptors(conf.getBrokerInterceptorsDirectory(),
conf.getNarExtractionDirectory());
- ImmutableMap.Builder<String, BrokerInterceptorWithClassLoader> builder = ImmutableMap.builder();
+ // Use LinkedHashMap as a temporary container to ensure insertion order
+ Map<String, BrokerInterceptorWithClassLoader> orderedInterceptorMap = new LinkedHashMap<>();
conf.getBrokerInterceptors().forEach(interceptorName -> {
@@ -80,7 +84,7 @@ public class BrokerInterceptors implements BrokerInterceptor {
try {
interceptor = BrokerInterceptorUtils.load(definition, conf.getNarExtractionDirectory());
if (interceptor != null) {
- builder.put(interceptorName, interceptor);
+ orderedInterceptorMap.put(interceptorName, interceptor);
}
log.info("Successfully loaded broker interceptor for name `{}`", interceptorName);
} catch (IOException e) {
@@ -89,9 +93,8 @@ public class BrokerInterceptors implements BrokerInterceptor {
}
});
- Map<String, BrokerInterceptorWithClassLoader> interceptors = builder.build();
- if (!interceptors.isEmpty()) {
- return new BrokerInterceptors(interceptors);
+ if (!orderedInterceptorMap.isEmpty()) {
+ 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 21c99f8196fe..2b00265b538d 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
@@ -268,7 +268,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) 08/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 57b493c512585b7ba3b4f64e7a99df7aed3b50ac
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 61b255e830db..becc4f165881 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
@@ -81,6 +81,8 @@ import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.BundlesData;
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;
@@ -290,6 +292,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");
@@ -457,7 +473,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..dea26eac9c02 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.trim().equals("");
}
}
}
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());
+ }
+}
(pulsar) 11/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fb093d27ec954daae20c00a2b247bfdc7eeb084d
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 | 46 +---
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 105 +++----
.../buffer/impl/TopicTransactionBuffer.java | 212 +++++++++++----
.../pulsar/broker/transaction/TransactionTest.java | 7 +-
.../AbstractSnapshotAbortedTxnProcessorTest.java | 302 +++++++++++++++++++++
...napshotSegmentAbortedTxnProcessorCloseTest.java | 173 ++++++++++++
.../impl/TopicTransactionBufferCloseTest.java | 283 +++++++++++++++++++
.../impl/TopicTransactionBufferRecoveryTest.java | 88 ++++++
9 files changed, 1264 insertions(+), 138 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..2c8e9acd4a1e
--- /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.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+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 ScheduledExecutorService 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(ScheduledExecutorService 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(ScheduledExecutorService 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 5d360a5822df..fb84be7f0bae 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
@@ -30,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;
@@ -40,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;
/**
@@ -51,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()
+ .chooseThread(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) -> {
@@ -89,29 +88,15 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
- final var future = new CompletableFuture<Position>();
+ Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
- scheduledExecutor.execute(() -> {
- try {
- final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
- .getTableView(scheduledExecutor).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(executor).readLatest(topic.getName());
+ if (isClosed() || snapshot == null) {
+ return null;
+ }
+ handleSnapshot(snapshot);
+ return PositionFactory.create(snapshot.getMaxReadPositionLedgerId(), snapshot.getMaxReadPositionEntryId());
}
@Override
@@ -171,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 8fa917ab7e6b..435c5762a867 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
@@ -49,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;
@@ -72,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.
@@ -138,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()
+ .chooseThread(TopicName.get(topic.getName()).getNamespace()));
this.topic = topic;
this.persistentWorker = new PersistentWorker(topic);
/*
@@ -229,48 +232,48 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
+ Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- final var future = new CompletableFuture<Position>();
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
- scheduledExecutor.execute(() -> {
- try {
- final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
- .getTxnBufferSnapshotIndexService().getTableView(scheduledExecutor)
- .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(executor)
+ .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 {
@@ -293,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())));
}
}
@@ -346,22 +350,24 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
// This method will be deprecated and removed in version 4.x.0
- private Position recoverOldSnapshot() throws Exception {
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = topic.getBrokerService().getPulsar()
- .getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
+ private Position recoverOldSnapshot(ScheduledExecutorService 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(scheduledExecutor).readLatest(topic.getName());
- if (snapshot == null) {
+ .getTableView(executor).readLatest(topic.getName());
+ if (isClosed() || snapshot == null) {
return null;
}
handleOldSnapshot(snapshot);
@@ -409,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 486a655e0187..adab16469b1c 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) {
@@ -189,8 +202,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);
}
@@ -199,14 +216,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) {
@@ -219,11 +236,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()
+ .chooseThread(TopicName.get(topic.getName()).getNamespace()).submit(replay);
+ }
+
@Override
public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
return CompletableFuture.completedFuture(null);
@@ -306,6 +337,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 =
@@ -431,7 +467,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);
}
@@ -461,6 +497,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
snapshotAbortedTxnProcessor.trimExpiredAbortedTxns();
takeSnapshotByChangeTimes();
}
+ updateLastDispatchablePosition(null);
txnCommittedCounter.increment();
completableFuture.complete(null);
}
@@ -506,8 +543,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);
}
}
@@ -598,8 +640,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);
}
/**
@@ -676,17 +716,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
@@ -790,8 +843,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();
@@ -812,7 +863,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
this.abortedTxnProcessor = abortedTxnProcessor;
}
- @SneakyThrows
@Override
public void run() {
if (!this.topicTransactionBuffer.changeToInitializingState()) {
@@ -820,30 +870,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) {
@@ -858,19 +940,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) {
@@ -926,6 +1008,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,
@@ -951,7 +1035,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
@@ -965,8 +1054,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 ee292750263d..425a62980fb9 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
@@ -37,6 +37,7 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
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;
@@ -69,6 +70,7 @@ import lombok.Cleanup;
import lombok.Lombok;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.util.Bytes;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -1584,7 +1586,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);
@@ -1638,6 +1640,9 @@ public class TransactionTest extends TransactionTestBase {
when(pulsar.getConfiguration()).thenReturn(serviceConfiguration);
when(pulsar.getConfig()).thenReturn(serviceConfiguration);
when(pulsar.getTransactionExecutorProvider()).thenReturn(executorProvider);
+ OrderedScheduler snapshotRecoverExecutorProvider = mock(OrderedScheduler.class);
+ when(snapshotRecoverExecutorProvider.chooseThread(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..5310a1099614
--- /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.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+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(ScheduledExecutorService recoveryExecutor) {
+ super(recoveryExecutor);
+ }
+
+ @Override
+ Position doRecoverFromSnapshot(ScheduledExecutorService 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..5f2196a2f13d
--- /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.common.util.OrderedScheduler;
+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.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);
+ OrderedScheduler recoveryScheduler = mock(OrderedScheduler.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.chooseThread(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(recoveryExecutor)).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..0ca3e19783f6
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
@@ -0,0 +1,283 @@
+/*
+ * 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.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
+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.buffer.Unpooled;
+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.common.util.OrderedScheduler;
+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.EntryImpl;
+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.apache.pulsar.common.protocol.Commands;
+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 = createTxnEntry(position, txnID, abort ? MarkerType.TXN_ABORT : MarkerType.TXN_COMMIT);
+ 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);
+ }
+ }
+
+ private EntryImpl createTxnEntry(Position position, TxnID txnID, MarkerType markerType) {
+ MessageMetadata messageMetadata = new MessageMetadata().setTxnidMostBits(txnID.getMostSigBits())
+ .setTxnidLeastBits(txnID.getLeastSigBits()).setMarkerType(markerType.getValue());
+ ByteBuf byteBuf =
+ serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, Unpooled.EMPTY_BUFFER);
+ return EntryImpl.create(position.getLedgerId(), position.getEntryId(), byteBuf);
+ }
+
+ @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);
+ OrderedScheduler recoveryScheduler = mock(OrderedScheduler.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.chooseThread(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..16bdfacc6c05 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().chooseThread(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) 07/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4505bf6484a1ccb084bc16dca1335d3fb8863b59
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 | 117 +++++++++++++++++++++
2 files changed, 121 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 34d90942f1c0..94a7eaf5505b 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
@@ -2294,7 +2294,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);
@@ -2339,9 +2340,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..b43c1e23981b
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReaderMessageTTLTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+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.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ReaderMessageTTLTest extends ProducerConsumerBase {
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+
+ @DataProvider
+ public Object[][] expiryPaths() {
+ return new Object[][] {{true}, {false}};
+ }
+
+ @Test(dataProvider = "expiryPaths", timeOut = 60000)
+ public void testReaderRetainsUnreadMessagesDuringExpiry(boolean sharedPosition) throws Exception {
+ String topicName = newTopicName();
+ String namespace = "my-property/my-ns";
+ admin.namespaces().setRetention(namespace, new RetentionPolicies(-1, -1));
+ admin.namespaces().setNamespaceMessageTTL(namespace, 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) pulsar.getBrokerService().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) 04/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 70a6914befb4eb6933e4524f9005839dd007400b
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 0c61aa9de503..f127793bb361 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
@@ -133,6 +140,19 @@ if [[ $JAVA_MAJOR_VERSION -ge 24 ]]; then
fi
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) 05/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 100ce94050c448e659fb584cbd4806a4aa7e3536
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 1b2a56933276..738d3315a4c5 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")
@@ -262,7 +267,7 @@ public class PerformanceClient extends CmdBase {
}
}
- SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket();
+ SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket(recorder);
try {
produceClient.start();
@@ -370,7 +375,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 "
@@ -441,7 +446,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;
@@ -452,8 +457,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) 03/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ea35cfe1037b5114a5126c13f84e7d18d3c01893
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 19b7792f8226..1b2a56933276 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/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a5e9037a433ac87d109531fa70d9168473931af9
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 db138989a8ee..d7553998d18f 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
@@ -31,6 +31,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;
@@ -39,7 +40,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;
@@ -113,6 +114,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) 09/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e3842dee2bc897112f5426559594baccc143b2ed
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 c8f6540f23fd..428e953f0a88 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
@@ -2486,8 +2486,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) 02/11: [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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 57375e1e3213c8acc7cf5bd2020a05aa4b8c4eb4
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 b894d794ac3f..52b83b2f90cb 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
@@ -129,6 +129,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
@@ -1742,6 +1749,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";