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/06/09 13:47:58 UTC
(pulsar) branch branch-3.0 updated (f50e94aa760 -> 1294a56b95b)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from f50e94aa760 [improve][offloaders] Automatically evict Offloaded Ledgers from memory (#19783)
new dcbe23f94a8 [improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
new 31bae699ef3 [improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
new 17f04efa17b [improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
new 5a7ad7edad4 [improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
new 9b60bd4cab4 [improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
new 1294a56b95b [fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
The 6 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 +-
conf/broker.conf | 57 ++++++++++++++++++----
conf/standalone.conf | 44 +++++++++++++++--
distribution/server/src/assemble/LICENSE.bin.txt | 54 ++++++++++----------
distribution/shell/src/assemble/LICENSE.bin.txt | 52 ++++++++++----------
.../bookkeeper/mledger/ManagedLedgerConfig.java | 10 ++++
.../bookkeeper/mledger/OffloadedLedgerHandle.java | 4 ++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 13 +++--
pom.xml | 2 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 24 +++++++--
.../pulsar/broker/service/BrokerService.java | 2 +
...onPersistentDispatcherSingleActiveConsumer.java | 2 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 2 +-
.../jcloud/impl/BlobStoreBackedReadHandleImpl.java | 24 +++++++--
14 files changed, 208 insertions(+), 84 deletions(-)
(pulsar) 04/06: [improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a7ad7edad4c095162839b4de46cc213eae95a99
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 10:20:27 2025 +0300
[improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
(cherry picked from commit 1cc2402e07d225276f81d891bd466e7d6f11cbb9)
---
conf/broker.conf | 23 +++++++++++++++++++++++
conf/standalone.conf | 23 +++++++++++++++++++++++
2 files changed, 46 insertions(+)
diff --git a/conf/broker.conf b/conf/broker.conf
index 0d942da8182..a5c15cce46b 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1267,6 +1267,29 @@ managedLedgerUnackedRangesOpenCacheSetEnabled=true
# MetadataStore.
managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
+# ManagedCursorInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is NONE, then save the ManagedCursorInfo bytes data directly without compression.
+# Using compression reduces the size of persistent cursor (subscription) metadata. This enables using a higher
+# managedLedgerMaxUnackedRangesToPersistInMetadataStore value and reduces the overall metadata stored in
+# the metadata store such as ZooKeeper.
+managedCursorInfoCompressionType=NONE
+
+# ManagedCursorInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedCursorInfoCompressionThresholdInBytes=16384
+
+# ManagedLedgerInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is invalid or NONE, then save the ManagedLedgerInfo bytes data directly without compression.
+# Using compression reduces the size of the persistent topic metadata. When a topic contains a large number of
+# individual ledgers in BookKeeper or tiered storage, compression helps prevent the metadata size from exceeding
+# the maximum size of a metadata store entry (ZNode in ZooKeeper). This also reduces the overall metadata stored
+# in the metadata store such as ZooKeeper.
+managedLedgerInfoCompressionType=NONE
+
+# ManagedLedgerInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedLedgerInfoCompressionThresholdInBytes=16384
+
# Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
# corrupted at bookkeeper and managed-cursor is stuck at that ledger.
autoSkipNonRecoverableData=false
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 32bbce18405..f112b301450 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -813,6 +813,29 @@ managedLedgerUnackedRangesOpenCacheSetEnabled=true
# MetadataStore.
managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
+# ManagedCursorInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is NONE, then save the ManagedCursorInfo bytes data directly without compression.
+# Using compression reduces the size of persistent cursor (subscription) metadata. This enables using a higher
+# managedLedgerMaxUnackedRangesToPersistInMetadataStore value and reduces the overall metadata stored in
+# the metadata store such as ZooKeeper.
+managedCursorInfoCompressionType=NONE
+
+# ManagedCursorInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedCursorInfoCompressionThresholdInBytes=16384
+
+# ManagedLedgerInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is invalid or NONE, then save the ManagedLedgerInfo bytes data directly without compression.
+# Using compression reduces the size of the persistent topic metadata. When a topic contains a large number of
+# individual ledgers in BookKeeper or tiered storage, compression helps prevent the metadata size from exceeding
+# the maximum size of a metadata store entry (ZNode in ZooKeeper). This also reduces the overall metadata stored
+# in the metadata store such as ZooKeeper.
+managedLedgerInfoCompressionType=NONE
+
+# ManagedLedgerInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedLedgerInfoCompressionThresholdInBytes=16384
+
# Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
# corrupted at bookkeeper and managed-cursor is stuck at that ledger.
autoSkipNonRecoverableData=false
(pulsar) 03/06: [improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 17f04efa17bb51764172aa68608e20aabd9e13e8
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 10:20:14 2025 +0300
[improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
(cherry picked from commit d1bca65f9b8907308efe69c887516d3c6b514a10)
---
conf/broker.conf | 34 +++++++++++++++-------
conf/standalone.conf | 21 +++++++++++--
.../bookkeeper/mledger/ManagedLedgerConfig.java | 10 +++++++
.../apache/pulsar/broker/ServiceConfiguration.java | 24 +++++++++++----
.../pulsar/broker/service/BrokerService.java | 2 ++
5 files changed, 73 insertions(+), 18 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 10f268e74a0..0d942da8182 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1228,6 +1228,14 @@ managedLedgerCursorMaxEntriesPerLedger=50000
# Max time before triggering a rollover on a cursor ledger
managedLedgerCursorRolloverTimeInSeconds=14400
+# Maximum amount of memory used hold data read from storage (or from the cache).
+# This mechanism prevents the broker to have too many concurrent
+# reads from storage and fall into Out of Memory errors in case
+# of multiple concurrent reads to multiple concurrent consumers.
+# Set 0 in order to disable the feature.
+#
+managedLedgerMaxReadsInFlightSizeInMB=0
+
# Max number of "acknowledgment holes" that are going to be persistently stored.
# When acknowledging out of order, a consumer will leave holes that are supposed
# to be quickly filled by acking all the messages. The information of which
@@ -1237,13 +1245,22 @@ managedLedgerCursorRolloverTimeInSeconds=14400
# crashes.
managedLedgerMaxUnackedRangesToPersist=10000
-# Maximum amount of memory used hold data read from storage (or from the cache).
-# This mechanism prevents the broker to have too many concurrent
-# reads from storage and fall into Out of Memory errors in case
-# of multiple concurrent reads to multiple concurrent consumers.
-# Set 0 in order to disable the feature.
-#
-managedLedgerMaxReadsInFlightSizeInMB=0
+# Maximum number of partially acknowledged batch messages per subscription that will have their batch
+# deleted indexes persisted. Batch deleted index state is handled when acknowledgmentAtBatchIndexLevelEnabled=true.
+# When this limit is exceeded, remaining batch message containing the batch deleted indexes will
+# only be tracked in memory. In case of broker restarts or load balancing events, the batch
+# deleted indexes will be cleared while redelivering the messages to consumers.
+managedLedgerMaxBatchDeletedIndexToPersist=10000
+
+# When storing acknowledgement state, choose a more compact serialization format that stores
+# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
+# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+managedLedgerPersistIndividualAckAsLongArray=false
+
+# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
+# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
+# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
# Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
@@ -1696,9 +1713,6 @@ narExtractionDirectory=
# Maximum prefetch rounds for ledger reading for offloading
managedLedgerOffloadPrefetchRounds=1
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
# For Amazon S3 ledger offload, AWS region
s3ManagedLedgerOffloadRegion=
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 7db477a05ef..32bbce18405 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -791,6 +791,23 @@ managedLedgerMaxSizePerLedgerMbytes=2048
# crashes.
managedLedgerMaxUnackedRangesToPersist=10000
+# Maximum number of partially acknowledged batch messages per subscription that will have their batch
+# deleted indexes persisted. Batch deleted index state is handled when acknowledgmentAtBatchIndexLevelEnabled=true.
+# When this limit is exceeded, remaining batch message containing the batch deleted indexes will
+# only be tracked in memory. In case of broker restarts or load balancing events, the batch
+# deleted indexes will be cleared while redelivering the messages to consumers.
+managedLedgerMaxBatchDeletedIndexToPersist=10000
+
+# When storing acknowledgement state, choose a more compact serialization format that stores
+# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
+# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+managedLedgerPersistIndividualAckAsLongArray=false
+
+# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
+# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
+# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
+
# Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
# MetadataStore.
@@ -825,9 +842,6 @@ managedLedgerMinimumBacklogEntriesForCaching=1000
# Maximum backlog entry difference to prevent caching entries that can't be reused.
managedLedgerMaxBacklogBetweenCursorsForCaching=1000
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
# Managed ledger prometheus stats latency rollover seconds (default: 60s)
managedLedgerPrometheusStatsLatencyRolloverSeconds=60
@@ -1296,3 +1310,4 @@ delayedDeliveryMaxNumBuckets=-1
# Whether retain null-key message during topic compaction
topicCompactionRetainNullKey=true
+
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index 90f20e010ac..70e093750b9 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -487,6 +487,16 @@ public class ManagedLedgerConfig {
return maxBatchDeletedIndexToPersist;
}
+ /**
+ * Set max batch deleted index that will be persisted and recovered.
+ *
+ * @param maxBatchDeletedIndexToPersist
+ * max batch deleted index that will be persisted and recovered.
+ */
+ public void setMaxBatchDeletedIndexToPersist(int maxBatchDeletedIndexToPersist) {
+ this.maxBatchDeletedIndexToPersist = maxBatchDeletedIndexToPersist;
+ }
+
public boolean isPersistentUnackedRangesWithMultipleEntriesEnabled() {
return persistentUnackedRangesWithMultipleEntriesEnabled;
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 206a3dc0aa7..ab47ba7ecc7 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2152,10 +2152,22 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ " will only be tracked in memory and messages will be redelivered in case of"
+ " crashes.")
private int managedLedgerMaxUnackedRangesToPersist = 10000;
- @FieldContext(
- category = CATEGORY_STORAGE_ML,
- doc = "Whether persist cursor ack stats as long arrays, which will compress the data and reduce GC rate")
+
+ @FieldContext(category = CATEGORY_STORAGE_ML,
+ doc = "Maximum number of partially acknowledged batch messages per subscription that will have their batch "
+ + "deleted indexes persisted. Batch deleted index state is handled when "
+ + "acknowledgmentAtBatchIndexLevelEnabled=true.\n\n"
+ + "When this limit is exceeded, remaining batch message containing the batch deleted indexes will "
+ + "only be tracked in memory. In case of broker restarts or load balancing events, the batch "
+ + "deleted indexes will be cleared while redelivering the messages to consumers.")
+ private int managedLedgerMaxBatchDeletedIndexToPersist = 10000;
+
+ @FieldContext(category = CATEGORY_STORAGE_ML,
+ doc = "When storing acknowledgement state, choose a more compact serialization format that stores"
+ + " individual acknowledgements as a bitmap which is serialized to an array of long values.\n\n"
+ + "NOTE: This setting requires managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.")
private boolean managedLedgerPersistIndividualAckAsLongArray = false;
+
@FieldContext(
category = CATEGORY_STORAGE_ML,
doc = "If enabled, the maximum \"acknowledgment holes\" will not be limited and \"acknowledgment holes\" "
@@ -2178,8 +2190,10 @@ public class ServiceConfiguration implements PulsarConfiguration {
private int managedLedgerMaxUnackedRangesToPersistInMetadataStore = 1000;
@FieldContext(
category = CATEGORY_STORAGE_OFFLOADING,
- doc = "Use Open Range-Set to cache unacked messages (it is memory efficient but it can take more cpu)"
- )
+ doc = "When set to true, a BitSet will be used to track acknowledged messages that come after the \"mark "
+ + "delete position\" for each subscription.\n\nRoaringBitmap is used as a memory efficient BitSet "
+ + "implementation for the acknowledged messages tracking. Unacknowledged ranges are the message "
+ + "ranges excluding the acknowledged messages.")
private boolean managedLedgerUnackedRangesOpenCacheSetEnabled = true;
@FieldContext(
dynamic = true,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index a19098edce9..e28aa51265f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1989,6 +1989,8 @@ public class BrokerService implements Closeable {
managedLedgerConfig
.setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist());
+ managedLedgerConfig.setMaxBatchDeletedIndexToPersist(
+ serviceConfig.getManagedLedgerMaxBatchDeletedIndexToPersist());
managedLedgerConfig
.setPersistIndividualAckAsLongArray(serviceConfig.isManagedLedgerPersistIndividualAckAsLongArray());
managedLedgerConfig.setPersistentUnackedRangesWithMultipleEntriesEnabled(
(pulsar) 05/06: [improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9b60bd4cab41ba5ceb73a0ba22bcf0ed866f7dd6
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 11:56:01 2025 +0300
[improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
(cherry picked from commit db459477dd2548bebbbd519af8ade9e7857fa0a8)
---
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 6f9db92a745..3076e2a9b6b 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.121.Final</netty.version>
+ <netty.version>4.1.122.Final</netty.version>
<guice.version>4.2.3</guice.version>
<guava.version>32.1.1-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 c9c5e99c5b8..f331f1f11eb 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -289,33 +289,33 @@ The Apache Software License, Version 2.0
- org.apache.commons-commons-lang3-3.11.jar
- org.apache.commons-commons-text-1.10.0.jar
* Netty
- - io.netty-netty-buffer-4.1.121.Final.jar
- - io.netty-netty-codec-4.1.121.Final.jar
- - io.netty-netty-codec-dns-4.1.121.Final.jar
- - io.netty-netty-codec-http-4.1.121.Final.jar
- - io.netty-netty-codec-http2-4.1.121.Final.jar
- - io.netty-netty-codec-socks-4.1.121.Final.jar
- - io.netty-netty-codec-haproxy-4.1.121.Final.jar
- - io.netty-netty-common-4.1.121.Final.jar
- - io.netty-netty-handler-4.1.121.Final.jar
- - io.netty-netty-handler-proxy-4.1.121.Final.jar
- - io.netty-netty-resolver-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-classes-macos-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.121.Final-osx-aarch_64.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.121.Final-osx-x86_64.jar
- - io.netty-netty-transport-4.1.121.Final.jar
- - io.netty-netty-transport-classes-epoll-4.1.121.Final.jar
- - io.netty-netty-transport-native-epoll-4.1.121.Final-linux-aarch_64.jar
- - io.netty-netty-transport-native-epoll-4.1.121.Final-linux-x86_64.jar
- - io.netty-netty-transport-native-unix-common-4.1.121.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-linux-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-linux-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-osx-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-osx-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-windows-x86_64.jar
- - io.netty-netty-tcnative-classes-2.0.70.Final.jar
+ - io.netty-netty-buffer-4.1.122.Final.jar
+ - io.netty-netty-codec-4.1.122.Final.jar
+ - io.netty-netty-codec-dns-4.1.122.Final.jar
+ - io.netty-netty-codec-http-4.1.122.Final.jar
+ - io.netty-netty-codec-http2-4.1.122.Final.jar
+ - io.netty-netty-codec-socks-4.1.122.Final.jar
+ - io.netty-netty-codec-haproxy-4.1.122.Final.jar
+ - io.netty-netty-common-4.1.122.Final.jar
+ - io.netty-netty-handler-4.1.122.Final.jar
+ - io.netty-netty-handler-proxy-4.1.122.Final.jar
+ - io.netty-netty-resolver-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-classes-macos-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.122.Final-osx-aarch_64.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.122.Final-osx-x86_64.jar
+ - io.netty-netty-transport-4.1.122.Final.jar
+ - io.netty-netty-transport-classes-epoll-4.1.122.Final.jar
+ - io.netty-netty-transport-native-epoll-4.1.122.Final-linux-aarch_64.jar
+ - io.netty-netty-transport-native-epoll-4.1.122.Final-linux-x86_64.jar
+ - io.netty-netty-transport-native-unix-common-4.1.122.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.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 02e0c68d67d..f3cff534aea 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -348,35 +348,35 @@ The Apache Software License, Version 2.0
- commons-text-1.10.0.jar
- commons-compress-1.26.0.jar
* Netty
- - netty-buffer-4.1.121.Final.jar
- - netty-codec-4.1.121.Final.jar
- - netty-codec-dns-4.1.121.Final.jar
- - netty-codec-http-4.1.121.Final.jar
- - netty-codec-socks-4.1.121.Final.jar
- - netty-codec-haproxy-4.1.121.Final.jar
- - netty-common-4.1.121.Final.jar
- - netty-handler-4.1.121.Final.jar
- - netty-handler-proxy-4.1.121.Final.jar
- - netty-resolver-4.1.121.Final.jar
- - netty-resolver-dns-4.1.121.Final.jar
- - netty-transport-4.1.121.Final.jar
- - netty-transport-classes-epoll-4.1.121.Final.jar
- - netty-transport-native-epoll-4.1.121.Final-linux-aarch_64.jar
- - netty-transport-native-epoll-4.1.121.Final-linux-x86_64.jar
- - netty-transport-native-unix-common-4.1.121.Final.jar
- - netty-tcnative-boringssl-static-2.0.70.Final.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-linux-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-linux-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-osx-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-osx-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-windows-x86_64.jar
- - netty-tcnative-classes-2.0.70.Final.jar
+ - netty-buffer-4.1.122.Final.jar
+ - netty-codec-4.1.122.Final.jar
+ - netty-codec-dns-4.1.122.Final.jar
+ - netty-codec-http-4.1.122.Final.jar
+ - netty-codec-socks-4.1.122.Final.jar
+ - netty-codec-haproxy-4.1.122.Final.jar
+ - netty-common-4.1.122.Final.jar
+ - netty-handler-4.1.122.Final.jar
+ - netty-handler-proxy-4.1.122.Final.jar
+ - netty-resolver-4.1.122.Final.jar
+ - netty-resolver-dns-4.1.122.Final.jar
+ - netty-transport-4.1.122.Final.jar
+ - netty-transport-classes-epoll-4.1.122.Final.jar
+ - netty-transport-native-epoll-4.1.122.Final-linux-aarch_64.jar
+ - netty-transport-native-epoll-4.1.122.Final-linux-x86_64.jar
+ - netty-transport-native-unix-common-4.1.122.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-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.121.Final.jar
- - netty-resolver-dns-native-macos-4.1.121.Final-osx-aarch_64.jar
- - netty-resolver-dns-native-macos-4.1.121.Final-osx-x86_64.jar
+ - netty-resolver-dns-classes-macos-4.1.122.Final.jar
+ - netty-resolver-dns-native-macos-4.1.122.Final-osx-aarch_64.jar
+ - netty-resolver-dns-native-macos-4.1.122.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 04f36977d41..1e34501e2a5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -144,7 +144,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.121.Final</netty.version>
+ <netty.version>4.1.122.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) 01/06: [improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dcbe23f94a83658649e4eb6111ce0f2e2f5e3cc0
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jun 5 17:48:22 2025 +0800
[improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
(cherry picked from commit 21406b594a05a6362afdf68bbe5d041a5907394d)
---
.../bookkeeper/mledger/OffloadedLedgerHandle.java | 4 ++++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 12 +++++++----
.../jcloud/impl/BlobStoreBackedReadHandleImpl.java | 24 +++++++++++++++++-----
3 files changed, 31 insertions(+), 9 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
index f45d115090f..64464187895 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
@@ -26,4 +26,8 @@ public interface OffloadedLedgerHandle {
default long lastAccessTimestamp() {
return -1;
}
+
+ default int getPendingRead() {
+ return 0;
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index d2e3a0d951b..e911932be9c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -2658,10 +2658,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ledgerCache.forEach((ledgerId, ledger) -> {
if (ledger.isDone() && !ledger.isCompletedExceptionally()) {
ReadHandle readHandle = ledger.join();
- if (readHandle instanceof OffloadedLedgerHandle) {
- long lastAccessTimestamp = ((OffloadedLedgerHandle) readHandle).lastAccessTimestamp();
- if (lastAccessTimestamp >= 0) {
- long delta = now - lastAccessTimestamp;
+ if (readHandle instanceof OffloadedLedgerHandle offloadedLedgerHandle) {
+ int pendingRead = offloadedLedgerHandle.getPendingRead();
+ if (pendingRead == 0) {
+ long delta = now - offloadedLedgerHandle.lastAccessTimestamp();
if (delta >= inactiveOffloadedLedgerEvictionTimeMs) {
log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)",
name, ledgerId, delta);
@@ -2671,6 +2671,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
"[{}] Offloaded ledger {} cannot be released ({} ms elapsed since last access)",
name, ledgerId, delta);
}
+ } else if (pendingRead < 0) {
+ log.error("[{}] Offloaded ledger {} went to a wrong state because its pending read is a"
+ + " negative value {}. Please raise an issue to https://github.com/apache/pulsar", name,
+ ledgerId, pendingRead);
}
}
}
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
index 1f2f901f514..84d9d118728 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
@@ -29,6 +29,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
@@ -56,6 +57,9 @@ import org.slf4j.LoggerFactory;
public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedgerHandle {
private static final Logger log = LoggerFactory.getLogger(BlobStoreBackedReadHandleImpl.class);
+ protected static final AtomicIntegerFieldUpdater<BlobStoreBackedReadHandleImpl> PENDING_READ_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(BlobStoreBackedReadHandleImpl.class, "pendingRead");
+
private final long ledgerId;
private final OffloadIndexBlock index;
private final BackedInputStream inputStream;
@@ -71,6 +75,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
private volatile State state = null;
+ private volatile int pendingRead;
+
private volatile long lastAccessTimestamp = System.currentTimeMillis();
private BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index,
@@ -122,9 +128,17 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
getId(), firstEntry, lastEntry, (1 + lastEntry - firstEntry));
}
CompletableFuture<LedgerEntries> promise = new CompletableFuture<>();
- touch();
+
+ // Ledger handles will be only marked idle when "pendingRead" is "0", it is not needed to update
+ // "lastAccessTimestamp" if "pendingRead" is larger than "0".
+ // Rather than update "lastAccessTimestamp" when starts a reading, updating it when a reading task is finished
+ // is better.
+ PENDING_READ_UPDATER.incrementAndGet(this);
+ promise.whenComplete((__, ex) -> {
+ PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
+ lastAccessTimestamp = System.currentTimeMillis();
+ });
executor.execute(() -> {
- touch();
if (state == State.Closed) {
log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
ledgerId, firstEntry, lastEntry);
@@ -213,7 +227,6 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
}
private void seekToEntry(long nextExpectedId) throws IOException {
- touch();
Long knownOffset = entryOffsetsCache.getIfPresent(ledgerId, nextExpectedId);
if (knownOffset != null) {
inputStream.seek(knownOffset);
@@ -324,7 +337,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
return lastAccessTimestamp;
}
- private void touch() {
- lastAccessTimestamp = System.currentTimeMillis();
+ @Override
+ public int getPendingRead() {
+ return PENDING_READ_UPDATER.get(this);
}
}
(pulsar) 06/06: [fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1294a56b95b46d9d1e5d583c55d418d6d433a32f
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Jun 9 21:33:46 2025 +0800
[fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
(cherry picked from commit e0d7faa087f7ae6895f139b89ef333c8211f71d3)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 ++-
.../mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index e911932be9c..e6d38b699a4 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -2660,7 +2660,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ReadHandle readHandle = ledger.join();
if (readHandle instanceof OffloadedLedgerHandle offloadedLedgerHandle) {
int pendingRead = offloadedLedgerHandle.getPendingRead();
- if (pendingRead == 0) {
+ long lastAccessTimestamp = offloadedLedgerHandle.lastAccessTimestamp();
+ if (lastAccessTimestamp >= 0 && pendingRead == 0) {
long delta = now - offloadedLedgerHandle.lastAccessTimestamp();
if (delta >= inactiveOffloadedLedgerEvictionTimeMs) {
log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)",
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
index 84d9d118728..916e33c3642 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
@@ -135,8 +135,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
// is better.
PENDING_READ_UPDATER.incrementAndGet(this);
promise.whenComplete((__, ex) -> {
- PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
lastAccessTimestamp = System.currentTimeMillis();
+ PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
});
executor.execute(() -> {
if (state == State.Closed) {
(pulsar) 02/06: [improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 31bae699ef38f22510f5ec26ed9d865558214037
Author: Malla Sandeep <sa...@gmail.com>
AuthorDate: Fri Jun 6 01:57:49 2025 +0530
[improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
(cherry picked from commit cb0fe5b0f65f09937e136eda9afeef13a7b08a03)
---
.../nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java | 2 +-
.../NonPersistentStickyKeyDispatcherMultipleConsumers.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
index 25e3e2894da..862f1620c13 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
@@ -52,7 +52,7 @@ public final class NonPersistentDispatcherSingleActiveConsumer extends AbstractD
}
@Override
- public void sendMessages(List<Entry> entries) {
+ public synchronized void sendMessages(List<Entry> entries) {
Consumer currentConsumer = ACTIVE_CONSUMER_UPDATER.get(this);
if (currentConsumer != null && currentConsumer.getAvailablePermits() > 0 && currentConsumer.isWritable()) {
SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
index 2cad253f96e..90a944da7d3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
@@ -127,7 +127,7 @@ public class NonPersistentStickyKeyDispatcherMultipleConsumers extends NonPersis
};
@Override
- public void sendMessages(List<Entry> entries) {
+ public synchronized void sendMessages(List<Entry> entries) {
if (entries.isEmpty()) {
return;
}