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

docker/pulsar/scripts/update-ini-from-env.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import os
+import sys
+import configparser
+import re
+
+def get_first_word(section_name):
+    # Split the section name by any non-word character and return the first word
+    return re.split(r'\W+', section_name)[0]
+
+def update_ini_file(ini_file_path, env_prefix):
+    # Read the existing INI file
+    config = configparser.ConfigParser()
+    config.read(ini_file_path)
+
+    # Flag to track if any updates were made
+    updated = False
+
+    # Iterate over environment variables
+    for key, value in os.environ.items():
+        if env_prefix and not key.startswith(env_prefix):
+            continue
+
+        stripped_key = key[len(env_prefix):] if env_prefix else key
+
+        # Iterate through sections
+        for section in config.sections():
+            first_word = get_first_word(section)
+            prefix = first_word + '_'
+            if stripped_key.startswith(prefix):
+                config.set(section, stripped_key[len(prefix):], value)
+                updated = True
+                break
+            elif config.has_option(section, stripped_key):
+                config.set(section, stripped_key, value)
+                updated = True
+                break
+
+    # Write the updated INI file only if there were updates
+    if updated:
+        with open(ini_file_path, 'w') as configfile:
+            config.write(configfile)
+
+if __name__ == "__main__":
+    if len(sys.argv) != 3:
+        print("Usage: python3 update-ini-from-env.py <path_to_ini_file> <env_prefix>")
+        sys.exit(1)
+
+    ini_file_path = sys.argv[1]
+    env_prefix = sys.argv[2]
+    update_ini_file(ini_file_path, env_prefix)

docker/pulsar/scripts/update-rocksdb-conf-from-env.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# allows tuning of RocksDB configuration via environment variables which were effective
+# before Pulsar 2.11 / BookKeeper 4.15 / https://github.com/apache/bookkeeper/pull/3056
+# the script should be applied to the `conf/entry_location_rocksdb.conf` file
+
+import os
+import sys
+import configparser
+
+# Constants for section keys
+DB_OPTIONS = "DBOptions"
+CF_OPTIONS = "CFOptions \"default\""
+TABLE_OPTIONS = "TableOptions/BlockBasedTable \"default\""
+
+def update_ini_file(ini_file_path):
+    config = configparser.ConfigParser()
+    config.read(ini_file_path)
+    updated = False
+
+    # Mapping of environment variables to INI sections and keys
+    env_to_ini_mapping = {
+        "dbStorage_rocksDB_logPath": (DB_OPTIONS, "log_path"),
+        "dbStorage_rocksDB_logLevel": (DB_OPTIONS, "info_log_level"),
+        "dbStorage_rocksDB_lz4CompressionEnabled": (CF_OPTIONS, "compression"),
+        "dbStorage_rocksDB_writeBufferSizeMB": (CF_OPTIONS, "write_buffer_size"),
+        "dbStorage_rocksDB_sstSizeInMB": (CF_OPTIONS, "target_file_size_base"),
+        "dbStorage_rocksDB_blockSize": (TABLE_OPTIONS, "block_size"),
+        "dbStorage_rocksDB_bloomFilterBitsPerKey": (TABLE_OPTIONS, "filter_policy"),
+        "dbStorage_rocksDB_blockCacheSize": (TABLE_OPTIONS, "block_cache"),
+        "dbStorage_rocksDB_numLevels": (CF_OPTIONS, "num_levels"),
+        "dbStorage_rocksDB_numFilesInLevel0": (CF_OPTIONS, "level0_file_num_compaction_trigger"),
+        "dbStorage_rocksDB_maxSizeInLevel1MB": (CF_OPTIONS, "max_bytes_for_level_base"),
+        "dbStorage_rocksDB_format_version": (TABLE_OPTIONS, "format_version")
+    }
+
+    # Type conversion functions
+    def mb_to_bytes(mb):
+        return str(int(mb) * 1024 * 1024)
+
+    def str_to_bool(value):
+        return True if value.lower() in ["true", "1", "yes"] else False
+
+    # Iterate over environment variables
+    for key, value in os.environ.items():
+        if key.startswith("PULSAR_PREFIX_"):
+            key = key[len("PULSAR_PREFIX_"):]
+
+        if key in env_to_ini_mapping:
+            section, option = env_to_ini_mapping[key]
+            if key in ["dbStorage_rocksDB_writeBufferSizeMB", "dbStorage_rocksDB_sstSizeInMB", "dbStorage_rocksDB_maxSizeInLevel1MB"]:
+                value = mb_to_bytes(value)
+            elif key == "dbStorage_rocksDB_lz4CompressionEnabled":
+                value = "kLZ4Compression" if str_to_bool(value) else "kNoCompression"
+            elif key == "dbStorage_rocksDB_bloomFilterBitsPerKey":
+                value = "rocksdb.BloomFilter:{}:false".format(value)
+            if config.get(section, option, fallback=None) != value:
+                config.set(section, option, value)
+                updated = True
+
+    # Write the updated INI file only if there were updates
+    if updated:
+        with open(ini_file_path, 'w') as configfile:
+            config.write(configfile)
+
+if __name__ == "__main__":
+    ini_file_path = sys.argv[1] if len(sys.argv) > 1 else "conf/entry_location_rocksdb.conf"
+    update_ini_file(ini_file_path)

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

@@ -4100,6 +4100,8 @@ public static ManagedLedgerException createManagedLedgerException(int bkErrorCod
     public static ManagedLedgerException createManagedLedgerException(Throwable t) {
         if (t instanceof org.apache.bookkeeper.client.api.BKException) {
             return createManagedLedgerException(((org.apache.bookkeeper.client.api.BKException) t).getCode());
+        } else if (t instanceof ManagedLedgerException) {
+            return (ManagedLedgerException) t;
         } else if (t instanceof CompletionException
                 && !(t.getCause() instanceof CompletionException) /* check to avoid stackoverlflow */) {
             return createManagedLedgerException(t.getCause());

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

@@ -79,7 +79,7 @@ public void invalidateEntriesBeforeTimestamp(long timestamp) {
     @Override
     public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader,
                                final AsyncCallbacks.ReadEntriesCallback callback, Object ctx) {
-        lh.readAsync(firstEntry, lastEntry).thenAcceptAsync(
+        ReadEntryUtils.readAsync(ml, lh, firstEntry, lastEntry).thenAcceptAsync(
                 ledgerEntries -> {
                     List<Entry> entries = new ArrayList<>();
                     long totalSize = 0;
@@ -107,7 +107,7 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole
     @Override
     public void asyncReadEntry(ReadHandle lh, PositionImpl position, AsyncCallbacks.ReadEntryCallback callback,
                                Object ctx) {
-        lh.readAsync(position.getEntryId(), position.getEntryId()).whenCompleteAsync(
+        ReadEntryUtils.readAsync(ml, lh, position.getEntryId(), position.getEntryId()).whenCompleteAsync(
                 (ledgerEntries, exception) -> {
                     if (exception != null) {
                         ml.invalidateLedgerHandle(lh);

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

@@ -248,7 +248,7 @@ private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEnt
             manager.mlFactoryMBean.recordCacheHit(cachedEntry.getLength());
             callback.readEntryComplete(cachedEntry, ctx);
         } else {
-            lh.readAsync(position.getEntryId(), position.getEntryId()).thenAcceptAsync(
+            ReadEntryUtils.readAsync(ml, lh, position.getEntryId(), position.getEntryId()).thenAcceptAsync(
                     ledgerEntries -> {
                         try {
                             Iterator<LedgerEntry> iterator = ledgerEntries.iterator();
@@ -428,7 +428,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
     CompletableFuture<List<EntryImpl>> readFromStorage(ReadHandle lh,
                                                        long firstEntry, long lastEntry, boolean shouldCacheEntry) {
         final int entriesToRead = (int) (lastEntry - firstEntry) + 1;
-        CompletableFuture<List<EntryImpl>> readResult = lh.readAsync(firstEntry, lastEntry)
+        CompletableFuture<List<EntryImpl>> readResult = ReadEntryUtils.readAsync(ml, lh, firstEntry, lastEntry)
                 .thenApply(
                         ledgerEntries -> {
                             requireNonNull(ml.getName());

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

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl.cache;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+
+class ReadEntryUtils {
+
+    static CompletableFuture<LedgerEntries> readAsync(ManagedLedger ml, ReadHandle handle, long firstEntry,
+                                                      long lastEntry) {
+        if (ml.getOptionalLedgerInfo(handle.getId()).isEmpty()) {
+            // The read handle comes from another managed ledger, in this case, we can only compare the entry range with
+            // the LAC of that read handle. Specifically, it happens when this method is called by a
+            // ReadOnlyManagedLedgerImpl object.
+            return handle.readAsync(firstEntry, lastEntry);
+        }
+        // Compare the entry range with the lastConfirmedEntry maintained by the managed ledger because the entry cache
+        // of `ShadowManagedLedgerImpl` reads entries via `ReadOnlyLedgerHandle`, which never updates `lastAddConfirmed`
+        final var lastConfirmedEntry = ml.getLastConfirmedEntry();
+        if (lastConfirmedEntry == null) {
+            return CompletableFuture.failedFuture(new ManagedLedgerException(
+                    "LastConfirmedEntry is null when reading ledger " + handle.getId()));
+        }
+        if (handle.getId() > lastConfirmedEntry.getLedgerId()) {
+            return CompletableFuture.failedFuture(new ManagedLedgerException("LastConfirmedEntry is "
+                    + lastConfirmedEntry + " when reading ledger " + handle.getId()));
+        }
+        if (handle.getId() == lastConfirmedEntry.getLedgerId() && lastEntry > lastConfirmedEntry.getEntryId()) {
+            return CompletableFuture.failedFuture(new ManagedLedgerException("LastConfirmedEntry is "
+                    + lastConfirmedEntry + " when reading entry " + lastEntry));
+        }
+        return handle.readUnconfirmedAsync(firstEntry, lastEntry);
+    }
+}

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java

@@ -28,6 +28,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -43,6 +44,7 @@
 import org.apache.bookkeeper.mledger.impl.cache.EntryCache;
 import org.apache.bookkeeper.mledger.impl.cache.EntryCacheDisabled;
 import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
 import org.testng.Assert;
 import org.testng.annotations.Test;
@@ -390,6 +392,9 @@ void entryCacheDisabledAsyncReadEntry() throws Exception {
         EntryCache entryCache = cacheManager.getEntryCache(ml1);
 
         final CountDownLatch counter = new CountDownLatch(1);
+        when(ml1.getLastConfirmedEntry()).thenReturn(new PositionImpl(1L, 1L));
+        when(ml1.getOptionalLedgerInfo(lh.getId())).thenReturn(Optional.of(mock(
+                MLDataFormats.ManagedLedgerInfo.LedgerInfo.class)));
         entryCache.asyncReadEntry(lh, new PositionImpl(1L,1L), new AsyncCallbacks.ReadEntryCallback() {
             public void readEntryComplete(Entry entry, Object ctx) {
                 Assert.assertNotEquals(entry, null);
@@ -404,7 +409,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
         }, null);
         counter.await();
 
-        verify(lh).readAsync(anyLong(), anyLong());
+        verify(lh).readUnconfirmedAsync(anyLong(), anyLong());
     }
 
 }