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 2025/09/08 18:53:19 UTC
(pulsar) branch branch-4.1 updated (4097ce09426 -> eb39530e711)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 4097ce09426 [feat][misc] upgrade oxia version to 0.6.2 (#24689)
new 3f510c6a027 [improve][ml] Improve cache insert performance by removing exists check since it's already covered by putIfAbsent (#24699)
new 9e6f7a130e2 [improve][client] PIP-407 Add newMessage with schema and transactions (#23942)
new 77d18bfcd37 [fix][broker] Fix NPE and annotate nullable return values for ManagedCursorContainer (#24706)
new ec0df9ccffe [improve][build] Disable javadoc build failure (#24594)
new eb39530e711 [fix][sec] Upgrade Netty to 4.1.127.Final to address CVEs (#24717)
The 5 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:
buildtools/pom.xml | 2 +-
distribution/server/src/assemble/LICENSE.bin.txt | 54 +++++++++++-----------
distribution/shell/src/assemble/LICENSE.bin.txt | 52 ++++++++++-----------
.../apache/bookkeeper/mledger/ManagedLedger.java | 4 +-
.../mledger/impl/ManagedCursorContainer.java | 17 +++++--
.../mledger/impl/cache/RangeEntryCacheImpl.java | 7 +--
pom.xml | 4 +-
.../pulsar/broker/service/BacklogQuotaManager.java | 3 ++
.../broker/service/persistent/PersistentTopic.java | 7 ++-
.../pulsar/broker/transaction/TransactionTest.java | 37 +++++++++++++++
.../org/apache/pulsar/client/api/Producer.java | 15 ++++++
.../apache/pulsar/client/impl/ProducerBase.java | 9 ++++
12 files changed, 144 insertions(+), 67 deletions(-)
(pulsar) 02/05: [improve][client] PIP-407 Add newMessage with schema and transactions (#23942)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9e6f7a130e22381a9726bf02ea985a9ab380a633
Author: Omar Yasin <om...@gmail.com>
AuthorDate: Thu Sep 4 23:11:31 2025 +0100
[improve][client] PIP-407 Add newMessage with schema and transactions (#23942)
Co-authored-by: Ómar Kjartan Yasin <oy...@apple.com>
(cherry picked from commit 536dce3a9bbb62985400acf69d67516603a300a8)
---
.../pulsar/broker/transaction/TransactionTest.java | 37 ++++++++++++++++++++++
.../org/apache/pulsar/client/api/Producer.java | 15 +++++++++
.../apache/pulsar/client/impl/ProducerBase.java | 9 ++++++
3 files changed, 61 insertions(+)
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 c54ffe9a73a..c3f969bf5f7 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
@@ -35,6 +35,7 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
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.fail;
import io.netty.buffer.Unpooled;
@@ -488,6 +489,42 @@ public class TransactionTest extends TransactionTestBase {
});
}
+ @Test
+ public void testSendAndAckWithSchema() throws Exception {
+ String topic = NAMESPACE1 + "/testAsyncSendAndAckWithSchema";
+ String topicName = "subscription";
+ getPulsarServiceList().get(0).getConfig().setBrokerDeduplicationEnabled(false);
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .producerName("producer")
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .create();
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName(topicName)
+ .subscribe();
+
+ Transaction txn = pulsarClient.newTransaction()
+ .withTransactionTimeout(10, TimeUnit.SECONDS)
+ .build()
+ .get();
+
+ MessageId messageId = producer.newMessage(Schema.STRING, txn)
+ .value("testSendAndAckWithSchema")
+ .send();
+
+ txn.commit().get();
+
+ Message<byte[]> message = consumer.receive();
+ assertEquals(new String(message.getValue(), StandardCharsets.UTF_8), "testSendAndAckWithSchema");
+ assertNotNull(message.getSchemaVersion());
+ }
+
@Test
public void testGetTxnID() throws Exception {
Transaction transaction = pulsarClient.newTransaction()
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java
index 3c5102eed2f..d6b07713708 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java
@@ -191,6 +191,21 @@ public interface Producer<T> extends Closeable {
* @since 2.7.0
*/
TypedMessageBuilder<T> newMessage(Transaction txn);
+
+ /**
+ * Create a new message builder with transaction and schema, not required same parameterized type with the
+ * producer.
+ *
+ * <p>After the transaction commit, it will be made visible to consumer.
+ *
+ * <p>After the transaction abort, it will never be visible to consumer.
+ *
+ * @return a typed message builder that can be used to construct the message to be sent through this producer
+ * @see #newMessage()
+ */
+ <V> TypedMessageBuilder<V> newMessage(Schema<V> schema,
+ Transaction txn);
+
/**
* Get the last sequence id that was published by this producer.
*
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
index 12e380fdd51..b7085d28ee2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
@@ -79,6 +79,7 @@ public abstract class ProducerBase<T> extends HandlerState implements Producer<T
return new TypedMessageBuilderImpl<>(this, schema);
}
+ @Override
public <V> TypedMessageBuilder<V> newMessage(Schema<V> schema) {
checkArgument(schema != null);
return new TypedMessageBuilderImpl<>(this, schema);
@@ -92,6 +93,14 @@ public abstract class ProducerBase<T> extends HandlerState implements Producer<T
return new TypedMessageBuilderImpl<>(this, schema, (TransactionImpl) txn);
}
+ @Override
+ public <V> TypedMessageBuilder<V> newMessage(Schema<V> schema,
+ Transaction txn) {
+ checkArgument(txn instanceof TransactionImpl);
+ checkArgument(schema != null);
+ return new TypedMessageBuilderImpl<>(this, schema, (TransactionImpl) txn);
+ }
+
abstract CompletableFuture<MessageId> internalSendAsync(Message<?> message);
abstract CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transaction txn);
(pulsar) 04/05: [improve][build] Disable javadoc build failure (#24594)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ec0df9ccffec9f11380dfddef949efa59293beb7
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Sep 8 22:22:17 2025 +0800
[improve][build] Disable javadoc build failure (#24594)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 783172fbe142e1a7e0d363a9a4e02ebff44f4141)
---
pom.xml | 2 ++
1 file changed, 2 insertions(+)
diff --git a/pom.xml b/pom.xml
index 4085367a0b0..22f4c6e0790 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2291,6 +2291,7 @@ flexible messaging model and an intuitive client API.</description>
<configuration>
<doclint>none</doclint>
<notimestamp>true</notimestamp>
+ <failOnError>false</failOnError>
</configuration>
</plugin>
<plugin>
@@ -2524,6 +2525,7 @@ flexible messaging model and an intuitive client API.</description>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<doclint>none</doclint>
+ <failOnError>false</failOnError>
</configuration>
</plugin>
<plugin>
(pulsar) 05/05: [fix][sec] Upgrade Netty to 4.1.127.Final to address CVEs (#24717)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eb39530e711f9326058535baca38f294fc365270
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Sep 8 21:49:12 2025 +0300
[fix][sec] Upgrade Netty to 4.1.127.Final to address CVEs (#24717)
(cherry picked from commit e9a9e04d7f9cb61ddaba056a31c97e04984b220c)
---
buildtools/pom.xml | 2 +-
distribution/server/src/assemble/LICENSE.bin.txt | 54 ++++++++++++------------
distribution/shell/src/assemble/LICENSE.bin.txt | 52 +++++++++++------------
pom.xml | 2 +-
4 files changed, 55 insertions(+), 55 deletions(-)
diff --git a/buildtools/pom.xml b/buildtools/pom.xml
index feb3f410414..4e7ae83e1ec 100644
--- a/buildtools/pom.xml
+++ b/buildtools/pom.xml
@@ -55,7 +55,7 @@
<license-maven-plugin.version>4.1</license-maven-plugin.version>
<puppycrawl.checkstyle.version>10.14.2</puppycrawl.checkstyle.version>
<maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
- <netty.version>4.1.124.Final</netty.version>
+ <netty.version>4.1.127.Final</netty.version>
<guice.version>4.2.3</guice.version>
<guava.version>33.4.8-jre</guava.version>
<ant.version>1.10.12</ant.version>
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 59691218908..46b62dde2b3 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -295,33 +295,33 @@ The Apache Software License, Version 2.0
- org.apache.commons-commons-lang3-3.18.0.jar
- org.apache.commons-commons-text-1.13.1.jar
* Netty
- - io.netty-netty-buffer-4.1.124.Final.jar
- - io.netty-netty-codec-4.1.124.Final.jar
- - io.netty-netty-codec-dns-4.1.124.Final.jar
- - io.netty-netty-codec-http-4.1.124.Final.jar
- - io.netty-netty-codec-http2-4.1.124.Final.jar
- - io.netty-netty-codec-socks-4.1.124.Final.jar
- - io.netty-netty-codec-haproxy-4.1.124.Final.jar
- - io.netty-netty-common-4.1.124.Final.jar
- - io.netty-netty-handler-4.1.124.Final.jar
- - io.netty-netty-handler-proxy-4.1.124.Final.jar
- - io.netty-netty-resolver-4.1.124.Final.jar
- - io.netty-netty-resolver-dns-4.1.124.Final.jar
- - io.netty-netty-resolver-dns-classes-macos-4.1.124.Final.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.124.Final-osx-aarch_64.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.124.Final-osx-x86_64.jar
- - io.netty-netty-transport-4.1.124.Final.jar
- - io.netty-netty-transport-classes-epoll-4.1.124.Final.jar
- - io.netty-netty-transport-native-epoll-4.1.124.Final-linux-aarch_64.jar
- - io.netty-netty-transport-native-epoll-4.1.124.Final-linux-x86_64.jar
- - io.netty-netty-transport-native-unix-common-4.1.124.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-linux-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-linux-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-osx-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-osx-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-windows-x86_64.jar
- - io.netty-netty-tcnative-classes-2.0.72.Final.jar
+ - io.netty-netty-buffer-4.1.127.Final.jar
+ - io.netty-netty-codec-4.1.127.Final.jar
+ - io.netty-netty-codec-dns-4.1.127.Final.jar
+ - io.netty-netty-codec-http-4.1.127.Final.jar
+ - io.netty-netty-codec-http2-4.1.127.Final.jar
+ - io.netty-netty-codec-socks-4.1.127.Final.jar
+ - io.netty-netty-codec-haproxy-4.1.127.Final.jar
+ - io.netty-netty-common-4.1.127.Final.jar
+ - io.netty-netty-handler-4.1.127.Final.jar
+ - io.netty-netty-handler-proxy-4.1.127.Final.jar
+ - io.netty-netty-resolver-4.1.127.Final.jar
+ - io.netty-netty-resolver-dns-4.1.127.Final.jar
+ - io.netty-netty-resolver-dns-classes-macos-4.1.127.Final.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.127.Final-osx-aarch_64.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.127.Final-osx-x86_64.jar
+ - io.netty-netty-transport-4.1.127.Final.jar
+ - io.netty-netty-transport-classes-epoll-4.1.127.Final.jar
+ - io.netty-netty-transport-native-epoll-4.1.127.Final-linux-aarch_64.jar
+ - io.netty-netty-transport-native-epoll-4.1.127.Final-linux-x86_64.jar
+ - io.netty-netty-transport-native-unix-common-4.1.127.Final.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final-linux-aarch_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final-linux-x86_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final-osx-aarch_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final-osx-x86_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.73.Final-windows-x86_64.jar
+ - io.netty-netty-tcnative-classes-2.0.73.Final.jar
- io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.26.Final.jar
- io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar
- io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index e21fdfeafcf..aeff29513dd 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -350,35 +350,35 @@ The Apache Software License, Version 2.0
- commons-collections-3.2.2.jar
- commons-configuration2-2.12.0.jar
* Netty
- - netty-buffer-4.1.124.Final.jar
- - netty-codec-4.1.124.Final.jar
- - netty-codec-dns-4.1.124.Final.jar
- - netty-codec-http-4.1.124.Final.jar
- - netty-codec-socks-4.1.124.Final.jar
- - netty-codec-haproxy-4.1.124.Final.jar
- - netty-common-4.1.124.Final.jar
- - netty-handler-4.1.124.Final.jar
- - netty-handler-proxy-4.1.124.Final.jar
- - netty-resolver-4.1.124.Final.jar
- - netty-resolver-dns-4.1.124.Final.jar
- - netty-transport-4.1.124.Final.jar
- - netty-transport-classes-epoll-4.1.124.Final.jar
- - netty-transport-native-epoll-4.1.124.Final-linux-aarch_64.jar
- - netty-transport-native-epoll-4.1.124.Final-linux-x86_64.jar
- - netty-transport-native-unix-common-4.1.124.Final.jar
- - netty-tcnative-boringssl-static-2.0.72.Final.jar
- - netty-tcnative-boringssl-static-2.0.72.Final-linux-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.72.Final-linux-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.72.Final-osx-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.72.Final-osx-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.72.Final-windows-x86_64.jar
- - netty-tcnative-classes-2.0.72.Final.jar
+ - netty-buffer-4.1.127.Final.jar
+ - netty-codec-4.1.127.Final.jar
+ - netty-codec-dns-4.1.127.Final.jar
+ - netty-codec-http-4.1.127.Final.jar
+ - netty-codec-socks-4.1.127.Final.jar
+ - netty-codec-haproxy-4.1.127.Final.jar
+ - netty-common-4.1.127.Final.jar
+ - netty-handler-4.1.127.Final.jar
+ - netty-handler-proxy-4.1.127.Final.jar
+ - netty-resolver-4.1.127.Final.jar
+ - netty-resolver-dns-4.1.127.Final.jar
+ - netty-transport-4.1.127.Final.jar
+ - netty-transport-classes-epoll-4.1.127.Final.jar
+ - netty-transport-native-epoll-4.1.127.Final-linux-aarch_64.jar
+ - netty-transport-native-epoll-4.1.127.Final-linux-x86_64.jar
+ - netty-transport-native-unix-common-4.1.127.Final.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final-linux-aarch_64.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final-linux-x86_64.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final-osx-aarch_64.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final-osx-x86_64.jar
+ - netty-tcnative-boringssl-static-2.0.73.Final-windows-x86_64.jar
+ - netty-tcnative-classes-2.0.73.Final.jar
- netty-incubator-transport-classes-io_uring-0.0.26.Final.jar
- netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar
- netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar
- - netty-resolver-dns-classes-macos-4.1.124.Final.jar
- - netty-resolver-dns-native-macos-4.1.124.Final-osx-aarch_64.jar
- - netty-resolver-dns-native-macos-4.1.124.Final-osx-x86_64.jar
+ - netty-resolver-dns-classes-macos-4.1.127.Final.jar
+ - netty-resolver-dns-native-macos-4.1.127.Final-osx-aarch_64.jar
+ - netty-resolver-dns-native-macos-4.1.127.Final-osx-x86_64.jar
* Prometheus client
- simpleclient-0.16.0.jar
- simpleclient_log4j2-0.16.0.jar
diff --git a/pom.xml b/pom.xml
index 22f4c6e0790..b397229ea37 100644
--- a/pom.xml
+++ b/pom.xml
@@ -189,7 +189,7 @@ flexible messaging model and an intuitive client API.</description>
<snappy.version>1.1.10.5</snappy.version> <!-- ZooKeeper server -->
<dropwizardmetrics.version>4.1.12.1</dropwizardmetrics.version> <!-- ZooKeeper server -->
<curator.version>5.7.1</curator.version>
- <netty.version>4.1.124.Final</netty.version>
+ <netty.version>4.1.127.Final</netty.version>
<netty-iouring.version>0.0.26.Final</netty-iouring.version>
<jetty.version>9.4.57.v20241219</jetty.version>
<conscrypt.version>2.5.2</conscrypt.version>
(pulsar) 03/05: [fix][broker] Fix NPE and annotate nullable return values for ManagedCursorContainer (#24706)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 77d18bfcd372509c3074beea024f1189581116ef
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Sep 5 15:54:14 2025 +0800
[fix][broker] Fix NPE and annotate nullable return values for ManagedCursorContainer (#24706)
(cherry picked from commit e6f66dfa016fcad74ca72808fe7654f1204a5920)
---
.../org/apache/bookkeeper/mledger/ManagedLedger.java | 4 +++-
.../bookkeeper/mledger/impl/ManagedCursorContainer.java | 17 +++++++++++++----
.../pulsar/broker/service/BacklogQuotaManager.java | 3 +++
.../broker/service/persistent/PersistentTopic.java | 7 ++++++-
4 files changed, 25 insertions(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index 20ff66aa70e..8fb083bcd02 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -38,6 +38,7 @@ import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
+import org.jspecify.annotations.Nullable;
/**
* A ManagedLedger it's a superset of a BookKeeper ledger concept.
@@ -579,8 +580,9 @@ public interface ManagedLedger {
/**
* Get the slowest consumer.
*
- * @return the slowest consumer
+ * @return the slowest consumer or null if there is no consumer
*/
+ @Nullable
ManagedCursor getSlowestConsumer();
/**
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainer.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainer.java
index cc740f5d37c..677b2a0a6aa 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainer.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainer.java
@@ -23,12 +23,17 @@ import lombok.experimental.UtilityClass;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.Position;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.classification.InterfaceAudience;
+import org.apache.pulsar.common.classification.InterfaceStability;
+import org.jspecify.annotations.Nullable;
/**
* Contains cursors for a ManagedLedger.
* <p>
* The goal is to be able to find out the slowest cursor and hence decide which is the oldest ledger we need to keep.
*/
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
public interface ManagedCursorContainer extends Iterable<ManagedCursor> {
/**
* Adds a cursor to the container with the specified position.
@@ -45,6 +50,7 @@ public interface ManagedCursorContainer extends Iterable<ManagedCursor> {
* @param name the name of the cursor
* @return the ManagedCursor if found, otherwise null
*/
+ @Nullable
ManagedCursor get(String name);
/**
@@ -63,29 +69,32 @@ public interface ManagedCursorContainer extends Iterable<ManagedCursor> {
* @param cursor the cursor to update the position for
* @param newPosition the updated position for the cursor
* @return a pair of positions, representing the previous slowest cursor and the new slowest cursor (after the
- * update).
+ * update) or null if the cursor does not exist.
*/
Pair<Position, Position> cursorUpdated(ManagedCursor cursor, Position newPosition);
/**
* Gets the position of the slowest cursor.
*
- * @return the position of the slowest cursor
+ * @return the position of the slowest cursor or null if there is no cursor
*/
+ @Nullable
Position getSlowestCursorPosition();
/**
* Gets the slowest cursor.
*
- * @return the slowest ManagedCursor
+ * @return the slowest ManagedCursor or null if there is no cursor
*/
+ @Nullable
ManagedCursor getSlowestCursor();
/**
* Gets the cursor's {@link CursorInfo} with the oldest position.
*
- * @return the CursorInfo containing the cursor and its position
+ * @return the CursorInfo containing the cursor and its position or null if there is no cursor
*/
+ @Nullable
CursorInfo getCursorWithOldestPosition();
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
index 689e8514078..5e811ff1dce 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
@@ -219,6 +219,9 @@ public class BacklogQuotaManager {
try {
for (; ; ) {
ManagedCursor slowestConsumer = mLedger.getSlowestConsumer();
+ if (slowestConsumer == null) {
+ break;
+ }
Position oldestPosition = slowestConsumer.getMarkDeletedPosition();
if (log.isDebugEnabled()) {
log.debug("[{}] slowest consumer mark delete position is [{}], read position is [{}]",
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 cc6cbb348f3..657f03d6b38 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
@@ -2166,7 +2166,12 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private void checkMessageExpiryWithSharedPosition(ManagedLedgerImpl ml, int messageTtlInSeconds) {
// Find the target position at one time, then expire all subscriptions and replicators.
- ManagedCursor cursor = ml.getCursors().getCursorWithOldestPosition().getCursor();
+ final var cursorWithOldestPosition = ml.getCursors().getCursorWithOldestPosition();
+ if (cursorWithOldestPosition == null) {
+ // Skip checking message expiry for topics without subscription
+ return;
+ }
+ ManagedCursor cursor = cursorWithOldestPosition.getCursor();
PersistentMessageFinder finder = new PersistentMessageFinder(topic, cursor, brokerService.getPulsar()
.getConfig().getManagedLedgerCursorResetLedgerCloseTimestampMaxClockSkewMillis());
// Find the target position.
(pulsar) 01/05: [improve][ml] Improve cache insert performance by removing exists check since it's already covered by putIfAbsent (#24699)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3f510c6a02706edc62ded53d3ca5ac2963d2b7a6
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Sep 4 12:29:28 2025 +0300
[improve][ml] Improve cache insert performance by removing exists check since it's already covered by putIfAbsent (#24699)
(cherry picked from commit 669ab6135c3f44d7ff8afbb4298343492bb6c328)
---
.../apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java | 7 +------
1 file changed, 1 insertion(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 44056bc44aa..04de8dde0ad 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -139,12 +139,6 @@ public class RangeEntryCacheImpl implements EntryCache {
entryLength);
}
- Position position = entry.getPosition();
- if (entries.exists(position)) {
- // If the entry is already in the cache, don't insert it again
- return false;
- }
-
ByteBuf cachedData;
if (copyEntries) {
cachedData = copyEntry(entry);
@@ -156,6 +150,7 @@ public class RangeEntryCacheImpl implements EntryCache {
cachedData = entry.getDataBuffer().retain();
}
+ Position position = entry.getPosition();
ReferenceCountedEntry cacheEntry =
EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler());
cachedData.release();