[improve][misc] Sync commits from apache into 3.1_ds

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>

conf/broker.conf

@@ -194,6 +194,10 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
+# If 'strictSubscriptionNameVerification' is true, the new subscription name can only contain (a-zA-Z_0-9) and these
+# special chars -=:.
+strictlyVerifySubscriptionName=false
+
 # Enable subscription auto creation if new consumer connected (disable auto creation with value false)
 allowAutoSubscriptionCreation=true
 
@@ -1233,6 +1237,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
@@ -1242,19 +1254,51 @@ 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
 # 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
@@ -1709,9 +1753,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=
 

conf/standalone.conf

@@ -798,11 +798,51 @@ 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.
 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
@@ -832,9 +872,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
 
@@ -1148,6 +1185,10 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
+# If 'strictSubscriptionNameVerification' is true, the new subscription name can only contain (a-zA-Z_0-9) and these
+# special chars -=:.
+strictlyVerifySubscriptionName=false
+
 # Enable subscription auto creation if new consumer connected (disable auto creation with value false)
 allowAutoSubscriptionCreation=true
 
@@ -1314,3 +1355,4 @@ disableBrokerInterceptors=true
 
 # Whether retain null-key message during topic compaction
 topicCompactionRetainNullKey=true
+

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

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -344,35 +344,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

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -487,6 +487,16 @@ public int getMaxBatchDeletedIndexToPersist() {
         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;
     }

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;
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -2644,10 +2644,11 @@ List<Long> internalEvictOffloadedLedgers() {
         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();
+                        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)",
                                        name, ledgerId, delta);
@@ -2656,6 +2657,10 @@ List<Long> internalEvictOffloadedLedgers() {
                            log.debug("[{}] 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);
                    }
                 }
             }