[improve] Handle PositionInfo that's too large to serialize as a single entry

buildtools/src/main/resources/pulsar/suppressions.xml

@@ -36,6 +36,7 @@
     <suppress checks=".*" files=".+[\\/]generated[\\/].+\.java"/>
     <suppress checks=".*" files=".+[\\/]generated-sources[\\/].+\.java"/>
     <suppress checks=".*" files=".+[\\/]generated-test-sources[\\/].+\.java"/>
+    <suppress checks=".*" files=".+PositionInfoUtils.java"/>
 
     <!-- suppress most all checks expect below-->
     <suppress checks="^(?!.*(UnusedImports|IllegalImport)).*$" files=".*[\\/]src[\\/]test[\\/].*"/>

conf/broker.conf

@@ -1941,10 +1941,17 @@ managedLedgerMaxUnackedRangesToPersistInZooKeeper=-1
 # persist, it will help to reduce the duplicates caused by the ack state that can not be fully persistent.
 dispatcherPauseOnAckStatePersistentEnabled=false
 
-# If enabled, the maximum "acknowledgment holes" will not be limited and "acknowledgment holes" are stored in
-# multiple entries.
+# If enabled, the maximum "acknowledgment holes" (as defined by managedLedgerMaxUnackedRangesToPersist)
+# can be stored in multiple entries, allowing the higher limits.
 persistentUnackedRangesWithMultipleEntriesEnabled=false
 
+# If persistentUnackedRangesWithMultipleEntriesEnabled, this sets maximum entry size for storage in bytes.
+persistentUnackedRangesMaxEntrySize=1048576
+
+# Set the compression type to use for cursor info.
+# Possible options are NONE, LZ4, ZLIB, ZSTD, SNAPPY
+cursorInfoCompressionType=NONE
+
 # Deprecated - Use managedLedgerCacheEvictionIntervalMs instead
 managedLedgerCacheEvictionFrequency=0
 

conf/standalone.conf

@@ -826,6 +826,17 @@ managedLedgerMaxUnackedRangesToPersist=10000
 # MetadataStore.
 managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
 
+# If enabled, the maximum "acknowledgment holes" (as defined by managedLedgerMaxUnackedRangesToPersist)
+# can be stored in multiple entries, allowing the higher limits.
+persistentUnackedRangesWithMultipleEntriesEnabled=false
+
+# If persistentUnackedRangesWithMultipleEntriesEnabled, this sets maximum entry size for storage in bytes.
+persistentUnackedRangesMaxEntrySize=1048576
+
+# Set the compression type to use for cursor info.
+# Possible options are NONE, LZ4, ZLIB, ZSTD, SNAPPY
+cursorInfoCompressionType=NONE
+
 # 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

docker/pulsar/Dockerfile

@@ -141,6 +141,8 @@ COPY --from=pulsar /pulsar /pulsar
 
 WORKDIR /pulsar
 ENV PATH=$PATH:$JAVA_HOME/bin:/pulsar/bin
+# Use musl libc library for RocksDB
+ENV ROCKSDB_MUSL_LIBC=true
 
 # The UID must be non-zero. Otherwise, it is arbitrary. No logic should rely on its specific value.
 ARG DEFAULT_USERNAME=pulsar

managed-ledger/pom.xml

@@ -191,6 +191,22 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>com.github.splunk.lightproto</groupId>
+        <artifactId>lightproto-maven-plugin</artifactId>
+        <version>${lightproto-maven-plugin.version}</version>
+        <configuration>
+          <singleOuterClass>true</singleOuterClass>
+          <classPrefix>Light</classPrefix>
+        </configuration>
+        <executions>
+          <execution>
+            <goals>
+              <goal>generate</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-checkstyle-plugin</artifactId>

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

@@ -45,7 +45,9 @@ public class ManagedLedgerConfig {
     private boolean createIfMissing = true;
     private int maxUnackedRangesToPersist = 10000;
     private int maxBatchDeletedIndexToPersist = 10000;
+    private String cursorInfoCompressionType = "NONE";
     private boolean persistentUnackedRangesWithMultipleEntriesEnabled = false;
+    private int persistentUnackedRangesMaxEntrySize = 1024 * 1024;
     private boolean deletionAtBatchIndexLevelEnabled = true;
     private int maxUnackedRangesToPersistInMetadataStore = 1000;
     private int maxEntriesPerLedger = 50000;
@@ -480,14 +482,49 @@ public int getMaxBatchDeletedIndexToPersist() {
         return maxBatchDeletedIndexToPersist;
     }
 
+    /**
+     * @return true if persistent unacked ranges with multiple entries enabled.
+     */
     public boolean isPersistentUnackedRangesWithMultipleEntriesEnabled() {
         return persistentUnackedRangesWithMultipleEntriesEnabled;
     }
 
+    /**
+     * If enabled, the maximum "acknowledgment holes" will be stored in multiple entries, allowing the higher limits.
+     * @param multipleEntriesEnabled
+     */
     public void setPersistentUnackedRangesWithMultipleEntriesEnabled(boolean multipleEntriesEnabled) {
         this.persistentUnackedRangesWithMultipleEntriesEnabled = multipleEntriesEnabled;
     }
 
+    /**
+     * @return max entry size for persistent unacked ranges.
+     */
+    public int getPersistentUnackedRangesMaxEntrySize() {
+        return persistentUnackedRangesMaxEntrySize;
+    }
+
+    /**
+     * If persistentUnackedRangesWithMultipleEntriesEnabled, this sets maximum entry size for storage in bytes.
+     */
+    public void setPersistentUnackedRangesMaxEntrySize(int persistentUnackedRangesMaxEntrySize) {
+        this.persistentUnackedRangesMaxEntrySize = persistentUnackedRangesMaxEntrySize;
+    }
+
+    /**
+     * @return compression type to use for cursor info.
+     */
+    public String getCursorInfoCompressionType() {
+        return cursorInfoCompressionType;
+    }
+
+    /**
+     * Set the compression type to use for cursor info.
+     */
+    public void setCursorInfoCompressionType(String cursorInfoCompressionType) {
+        this.cursorInfoCompressionType = cursorInfoCompressionType;
+    }
+
     /**
      * @param maxUnackedRangesToPersist
      *            max unacked message ranges that will be persisted and receverd.

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

@@ -108,6 +108,16 @@ public class ManagedLedgerFactoryConfig {
      */
     private long managedCursorInfoCompressionThresholdInBytes = 0;
 
+    /**
+     * If enabled, the maximum "acknowledgment holes" will be stored in multiple entries, allowing the higher limits.
+     */
+    private boolean persistentUnackedRangesWithMultipleEntriesEnabled = false;
+
+    /**
+     * If persistentUnackedRangesWithMultipleEntriesEnabled, this sets maximum entry size for storage in bytes.
+     */
+    private int persistentUnackedRangesMaxEntrySize = 1024 * 1024;
+
     public MetadataCompressionConfig getCompressionConfigForManagedLedgerInfo() {
         return new MetadataCompressionConfig(managedLedgerInfoCompressionType,
                 managedLedgerInfoCompressionThresholdInBytes);

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

@@ -44,6 +44,7 @@ public final class LedgerMetadataUtils {
 
     private static final String METADATA_PROPERTY_MANAGED_LEDGER_NAME = "pulsar/managed-ledger";
     private static final String METADATA_PROPERTY_CURSOR_NAME = "pulsar/cursor";
+    public static final String METADATA_PROPERTY_CURSOR_COMPRESSION_TYPE = "pulsar/cursor-compressionType";
     private static final String METADATA_PROPERTY_COMPACTEDTOPIC = "pulsar/compactedTopic";
     private static final String METADATA_PROPERTY_COMPACTEDTO = "pulsar/compactedTo";
     private static final String METADATA_PROPERTY_SCHEMAID = "pulsar/schemaId";
@@ -72,8 +73,13 @@ static Map<String, byte[]> buildBaseManagedLedgerMetadata(String name) {
      * @return an immutable map which describes the cursor
      * @see #buildBaseManagedLedgerMetadata(java.lang.String)
      */
-    static Map<String, byte[]> buildAdditionalMetadataForCursor(String name) {
-        return Map.of(METADATA_PROPERTY_CURSOR_NAME, name.getBytes(StandardCharsets.UTF_8));
+    static Map<String, byte[]> buildAdditionalMetadataForCursor(String name, String compressionType) {
+        if (compressionType != null) {
+            return Map.of(METADATA_PROPERTY_CURSOR_NAME, name.getBytes(StandardCharsets.UTF_8),
+                    METADATA_PROPERTY_CURSOR_COMPRESSION_TYPE, compressionType.getBytes(StandardCharsets.UTF_8));
+        } else {
+            return Map.of(METADATA_PROPERTY_CURSOR_NAME, name.getBytes(StandardCharsets.UTF_8));
+        }
     }
 
     /**