[improve][broker] Sync commits from apache/pulsar branch-3.0 into 3.1_ds

.github/workflows/ci-go-functions.yaml

@@ -76,7 +76,7 @@ jobs:
     runs-on: ubuntu-22.04
     strategy:
       matrix:
-        go-version: [1.15, 1.16, 1.17]
+        go-version: ['1.21']
 
     steps:
       - name: Check out code into the Go module directory
@@ -94,7 +94,7 @@ jobs:
       - name: InstallTool
         run: |
           cd pulsar-function-go
-          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.18.0
+          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.55.2
           ./bin/golangci-lint --version
 
       - name: Build

bouncy-castle/bc/LICENSE

@@ -205,6 +205,6 @@
 This projects includes binary packages with the following licenses:
 Bouncy Castle License
  * Bouncy Castle -- licenses/LICENSE-bouncycastle.txt
-    - org.bouncycastle-bcpkix-jdk18on-1.75.jar
-    - org.bouncycastle-bcprov-jdk18on-1.75.jar
-    - org.bouncycastle-bcprov-ext-jdk18on-1.75.jar
+    - org.bouncycastle-bcpkix-jdk18on-1.78.jar
+    - org.bouncycastle-bcprov-jdk18on-1.78.jar
+    - org.bouncycastle-bcprov-ext-jdk18on-1.78.jar

conf/broker.conf

@@ -1148,6 +1148,16 @@ managedLedgerCacheEvictionTimeThresholdMillis=1000
 # and thus should be set as inactive.
 managedLedgerCursorBackloggedThreshold=1000
 
+# Minimum cursors that must be in backlog state to cache and reuse the read entries.
+# (Default =0 to disable backlog reach cache)
+managedLedgerMinimumBacklogCursorsForCaching=0
+
+# Minimum backlog entries for any cursor before start caching reads.
+managedLedgerMinimumBacklogEntriesForCaching=1000
+
+# Maximum backlog entry difference to prevent caching entries that can't be reused.
+managedLedgerMaxBacklogBetweenCursorsForCaching=1000
+
 # Rate limit the amount of writes per second generated by consumer acking the messages
 managedLedgerDefaultMarkDeleteRateLimit=1.0
 
@@ -1850,4 +1860,4 @@ brokerInterceptorsDirectory=./interceptors
 brokerInterceptors=
 
 # Enable or disable the broker interceptor, which is only used for testing for now
-disableBrokerInterceptors=true
+disableBrokerInterceptors=true

conf/proxy.conf

@@ -376,5 +376,7 @@ zooKeeperCacheExpirySeconds=-1
 enableProxyStatsEndpoints=true
 # Whether the '/metrics' endpoint requires authentication. Defaults to true
 authenticateMetricsEndpoint=true
-# Enable cache metrics data, default value is false
-metricsBufferResponse=false
+# Time in milliseconds that metrics endpoint would time out. Default is 30s.
+# Set it to 0 to disable timeout.
+metricsServletTimeoutMs=30000
+

conf/standalone.conf

@@ -813,10 +813,10 @@ managedLedgerNewEntriesCheckDelayInMillis=10
 managedLedgerMinimumBacklogCursorsForCaching=0
 
 # Minimum backlog entries for any cursor before start caching reads.
-managedLedgerMinimumBacklogEntriesForCaching=100
+managedLedgerMinimumBacklogEntriesForCaching=1000
 
 # Maximum backlog entry difference to prevent caching entries that can't be reused.
-managedLedgerMaxBacklogBetweenCursorsForCaching=10000
+managedLedgerMaxBacklogBetweenCursorsForCaching=1000
 
 # Use Open Range-Set to cache unacked messages
 managedLedgerUnackedRangesOpenCacheSetEnabled=true

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

@@ -345,34 +345,34 @@ The Apache Software License, Version 2.0
     - net.java.dev.jna-jna-jpms-5.12.1.jar
     - net.java.dev.jna-jna-platform-jpms-5.12.1.jar
  * BookKeeper
-    - org.apache.bookkeeper-bookkeeper-common-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-common-allocator-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-proto-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-server-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-tools-framework-4.16.4.jar
-    - org.apache.bookkeeper-circe-checksum-4.16.4.jar
-    - org.apache.bookkeeper-cpu-affinity-4.16.4.jar
-    - org.apache.bookkeeper-statelib-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-api-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-common-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-java-client-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-java-client-base-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-proto-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-server-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-service-api-4.16.4.jar
-    - org.apache.bookkeeper-stream-storage-service-impl-4.16.4.jar
-    - org.apache.bookkeeper.http-http-server-4.16.4.jar
-    - org.apache.bookkeeper.http-vertx-http-server-4.16.4.jar
-    - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.16.4.jar
-    - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.16.4.jar
-    - org.apache.distributedlog-distributedlog-common-4.16.4.jar
-    - org.apache.distributedlog-distributedlog-core-4.16.4-tests.jar
-    - org.apache.distributedlog-distributedlog-core-4.16.4.jar
-    - org.apache.distributedlog-distributedlog-protocol-4.16.4.jar
-    - org.apache.bookkeeper.stats-codahale-metrics-provider-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-slogger-api-4.16.4.jar
-    - org.apache.bookkeeper-bookkeeper-slogger-slf4j-4.16.4.jar
-    - org.apache.bookkeeper-native-io-4.16.4.jar
+    - org.apache.bookkeeper-bookkeeper-common-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-common-allocator-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-proto-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-server-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-tools-framework-4.16.5.jar
+    - org.apache.bookkeeper-circe-checksum-4.16.5.jar
+    - org.apache.bookkeeper-cpu-affinity-4.16.5.jar
+    - org.apache.bookkeeper-statelib-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-api-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-common-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-java-client-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-java-client-base-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-proto-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-server-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-service-api-4.16.5.jar
+    - org.apache.bookkeeper-stream-storage-service-impl-4.16.5.jar
+    - org.apache.bookkeeper.http-http-server-4.16.5.jar
+    - org.apache.bookkeeper.http-vertx-http-server-4.16.5.jar
+    - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.16.5.jar
+    - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.16.5.jar
+    - org.apache.distributedlog-distributedlog-common-4.16.5.jar
+    - org.apache.distributedlog-distributedlog-core-4.16.5-tests.jar
+    - org.apache.distributedlog-distributedlog-core-4.16.5.jar
+    - org.apache.distributedlog-distributedlog-protocol-4.16.5.jar
+    - org.apache.bookkeeper.stats-codahale-metrics-provider-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-slogger-api-4.16.5.jar
+    - org.apache.bookkeeper-bookkeeper-slogger-slf4j-4.16.5.jar
+    - org.apache.bookkeeper-native-io-4.16.5.jar
   * Apache HTTP Client
     - org.apache.httpcomponents-httpclient-4.5.13.jar
     - org.apache.httpcomponents-httpcore-4.4.15.jar
@@ -579,10 +579,10 @@ Creative Commons Attribution License
 
 Bouncy Castle License
  * Bouncy Castle -- ../licenses/LICENSE-bouncycastle.txt
-    - org.bouncycastle-bcpkix-jdk18on-1.75.jar
-    - org.bouncycastle-bcprov-ext-jdk18on-1.75.jar
-    - org.bouncycastle-bcprov-jdk18on-1.75.jar
-    - org.bouncycastle-bcutil-jdk18on-1.75.jar
+    - org.bouncycastle-bcpkix-jdk18on-1.78.jar
+    - org.bouncycastle-bcprov-ext-jdk18on-1.78.jar
+    - org.bouncycastle-bcprov-jdk18on-1.78.jar
+    - org.bouncycastle-bcutil-jdk18on-1.78.jar
 
 Datastax
     - com.datastax.oss-managed-ledger-3.1.3.2-SNAPSHOT.jar

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

@@ -386,9 +386,9 @@ The Apache Software License, Version 2.0
     - log4j-web-2.18.0.jar
 
  * BookKeeper
-    - bookkeeper-common-allocator-4.16.4.jar
-    - cpu-affinity-4.16.4.jar
-    - circe-checksum-4.16.4.jar
+    - bookkeeper-common-allocator-4.16.5.jar
+    - cpu-affinity-4.16.5.jar
+    - circe-checksum-4.16.5.jar
   * AirCompressor
      - aircompressor-0.20.jar
  * AsyncHttpClient
@@ -466,10 +466,10 @@ Creative Commons Attribution License
 
 Bouncy Castle License
  * Bouncy Castle -- ../licenses/LICENSE-bouncycastle.txt
-    - bcpkix-jdk18on-1.75.jar
-    - bcprov-ext-jdk18on-1.75.jar
-    - bcprov-jdk18on-1.75.jar
-    - bcutil-jdk18on-1.75.jar
+    - bcpkix-jdk18on-1.78.jar
+    - bcprov-ext-jdk18on-1.78.jar
+    - bcprov-jdk18on-1.78.jar
+    - bcutil-jdk18on-1.78.jar
 
 ------------------------
 

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

@@ -823,6 +823,11 @@ public void readEntryComplete(Entry entry, Object ctx) {
                 result.entry = entry;
                 counter.countDown();
             }
+
+            @Override
+            public String toString() {
+                return String.format("Cursor [%s] get Nth entry", ManagedCursorImpl.this);
+            }
         }, null);
 
         counter.await(ledger.getConfig().getMetadataOperationsTimeoutSeconds(), TimeUnit.SECONDS);
@@ -980,13 +985,18 @@ private void checkForNewEntries(OpReadEntry op, ReadEntriesCallback callback, Ob
                 log.debug("[{}] [{}] Re-trying the read at position {}", ledger.getName(), name, op.readPosition);
             }
 
+            if (isClosed()) {
+                callback.readEntriesFailed(new CursorAlreadyClosedException("Cursor was already closed"), ctx);
+                return;
+            }
+
             if (!hasMoreEntries()) {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}] Still no entries available. Register for notification", ledger.getName(),
                             name);
                 }
                 // Let the managed ledger know we want to be notified whenever a new entry is published
-                ledger.waitingCursors.add(this);
+                ledger.addWaitingCursor(this);
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}] Skip notification registering since we do have entries available",
@@ -1265,12 +1275,13 @@ protected void internalResetCursor(PositionImpl proposedReadPosition,
         if (proposedReadPosition.equals(PositionImpl.EARLIEST)) {
             newReadPosition = ledger.getFirstPosition();
         } else if (proposedReadPosition.equals(PositionImpl.LATEST)) {
-            newReadPosition = ledger.getLastPosition().getNext();
+            newReadPosition = ledger.getNextValidPosition(ledger.getLastPosition());
         } else {
             newReadPosition = proposedReadPosition;
         }
 
-        log.info("[{}] Initiate reset readPosition to {} on cursor {}", ledger.getName(), newReadPosition, name);
+        log.info("[{}] Initiate reset readPosition from {} to {} on cursor {}", ledger.getName(), readPosition,
+                newReadPosition, name);
 
         synchronized (pendingMarkDeleteOps) {
             if (!RESET_CURSOR_IN_PROGRESS_UPDATER.compareAndSet(this, FALSE, TRUE)) {
@@ -1536,6 +1547,11 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx)
                     callback.readEntriesFailed(exception.get(), ctx);
                 }
             }
+
+            @Override
+            public String toString() {
+                return String.format("Cursor [%s] async replay entries", ManagedCursorImpl.this);
+            }
         };
 
         positions.stream().filter(position -> !alreadyAcknowledgedPositions.contains(position))

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

@@ -737,6 +737,7 @@ public void operationComplete(MLDataFormats.ManagedLedgerInfo pbInfo, Stat stat)
                     ledgerInfo.ledgerId = pbLedgerInfo.getLedgerId();
                     ledgerInfo.entries = pbLedgerInfo.hasEntries() ? pbLedgerInfo.getEntries() : null;
                     ledgerInfo.size = pbLedgerInfo.hasSize() ? pbLedgerInfo.getSize() : null;
+                    ledgerInfo.timestamp = pbLedgerInfo.hasTimestamp() ? pbLedgerInfo.getTimestamp() : null;
                     ledgerInfo.isOffloaded = pbLedgerInfo.hasOffloadContext();
                     if (pbLedgerInfo.hasOffloadContext()) {
                         MLDataFormats.OffloadContext offloadContext = pbLedgerInfo.getOffloadContext();