[improve][broker] LS-1263: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

@@ -88,6 +88,16 @@ advertisedAddress=
 # If true, the real IP addresses of consumers and producers can be obtained when getting topic statistics data.
 haProxyProtocolEnabled=false
 
+# Enable or disable the use of HA proxy protocol for resolving the client IP for http/https requests.
+webServiceHaProxyProtocolEnabled=false
+
+# Trust X-Forwarded-For header for resolving the client IP for http/https requests. Default is false.
+webServiceTrustXForwardedFor=false
+
+# Add detailed client/remote and server/local addresses and ports to http/https request logging.
+# Defaults to true when either webServiceHaProxyProtocolEnabled or webServiceTrustXForwardedFor is enabled.
+webServiceLogDetailedAddresses=
+
 # Number of threads to config Netty Acceptor. Default is 1
 numAcceptorThreads=
 
@@ -1148,6 +1158,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
 
@@ -1651,10 +1671,10 @@ s3ManagedLedgerOffloadBucket=
 # For Amazon S3 ledger offload, Alternative endpoint to connect to (useful for testing)
 s3ManagedLedgerOffloadServiceEndpoint=
 
-# For Amazon S3 ledger offload, Max block size in bytes. (64MB by default, 5MB minimum)
+# For Amazon S3 ledger offload, Max block size in bytes. (64MiB by default, 5MiB minimum)
 s3ManagedLedgerOffloadMaxBlockSizeInBytes=67108864
 
-# For Amazon S3 ledger offload, Read buffer size in bytes (1MB by default)
+# For Amazon S3 ledger offload, Read buffer size in bytes (1MiB by default)
 s3ManagedLedgerOffloadReadBufferSizeInBytes=1048576
 
 # For Google Cloud Storage ledger offload, region where offload bucket is located.
@@ -1664,10 +1684,11 @@ gcsManagedLedgerOffloadRegion=
 # For Google Cloud Storage ledger offload, Bucket to place offloaded ledger into
 gcsManagedLedgerOffloadBucket=
 
-# For Google Cloud Storage ledger offload, Max block size in bytes. (64MB by default, 5MB minimum)
-gcsManagedLedgerOffloadMaxBlockSizeInBytes=67108864
+# For Google Cloud Storage ledger offload, Max block size in bytes. (128MiB by default, 5MiB minimum)
+# Since JClouds limits the maximum number of blocks to 32, the maximum size of a ledger is 32 times the block size.
+gcsManagedLedgerOffloadMaxBlockSizeInBytes=134217728
 
-# For Google Cloud Storage ledger offload, Read buffer size in bytes (1MB by default)
+# For Google Cloud Storage ledger offload, Read buffer size in bytes (1MiB by default)
 gcsManagedLedgerOffloadReadBufferSizeInBytes=1048576
 
 # For Google Cloud Storage, path to json file containing service account credentials.
@@ -1850,4 +1871,4 @@ brokerInterceptorsDirectory=./interceptors
 brokerInterceptors=
 
 # Enable or disable the broker interceptor, which is only used for testing for now
-disableBrokerInterceptors=true
+disableBrokerInterceptors=true

conf/functions_worker.yml

@@ -27,6 +27,16 @@ workerHostname: localhost
 workerPort: 6750
 workerPortTls: 6751
 
+# Enable or disable the use of HA proxy protocol for resolving the client IP for http/https requests.
+webServiceHaProxyProtocolEnabled: false
+
+# Trust X-Forwarded-For header for resolving the client IP for http/https requests. Default is false.
+webServiceTrustXForwardedFor: false
+
+# Add detailed client/remote and server/local addresses and ports to http/https request logging.
+# Defaults to true when either webServiceHaProxyProtocolEnabled or webServiceTrustXForwardedFor is enabled.
+webServiceLogDetailedAddresses: null
+
 # The Configuration metadata store url
 # Examples:
 # * zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181

conf/proxy.conf

@@ -63,6 +63,16 @@ advertisedAddress=
 # If true, the real IP addresses of consumers and producers can be obtained when getting topic statistics data.
 haProxyProtocolEnabled=false
 
+# Enable or disable the use of HA proxy protocol for resolving the client IP for http/https requests.
+webServiceHaProxyProtocolEnabled=false
+
+# Trust X-Forwarded-For header for resolving the client IP for http/https requests. Default is false.
+webServiceTrustXForwardedFor=false
+
+# Add detailed client/remote and server/local addresses and ports to http/https request logging.
+# Defaults to true when either webServiceHaProxyProtocolEnabled or webServiceTrustXForwardedFor is enabled.
+webServiceLogDetailedAddresses=
+
 # Enables zero-copy transport of data across network interfaces using the splice system call.
 # Zero copy mode cannot be used when TLS is enabled or when proxyLogLevel is > 0.
 proxyZeroCopyModeEnabled=true
@@ -376,5 +386,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

@@ -51,6 +51,16 @@ advertisedAddress=
 # If true, the real IP addresses of consumers and producers can be obtained when getting topic statistics data.
 haProxyProtocolEnabled=false
 
+# Enable or disable the use of HA proxy protocol for resolving the client IP for http/https requests.
+webServiceHaProxyProtocolEnabled=false
+
+# Trust X-Forwarded-For header for resolving the client IP for http/https requests. Default is false.
+webServiceTrustXForwardedFor=false
+
+# Add detailed client/remote and server/local addresses and ports to http/https request logging.
+# Defaults to true when either webServiceHaProxyProtocolEnabled or webServiceTrustXForwardedFor is enabled.
+webServiceLogDetailedAddresses=
+
 # Number of threads to use for Netty IO. Default is set to 2 * Runtime.getRuntime().availableProcessors()
 numIOThreads=
 
@@ -813,10 +823,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

conf/websocket.conf

@@ -46,6 +46,16 @@ statusFilePath=
 # Hostname or IP address the service binds on, default is 0.0.0.0.
 bindAddress=0.0.0.0
 
+# Enable or disable the use of HA proxy protocol for resolving the client IP for http/https requests.
+webServiceHaProxyProtocolEnabled=false
+
+# Trust X-Forwarded-For header for resolving the client IP for http/https requests. Default is false.
+webServiceTrustXForwardedFor=false
+
+# Add detailed client/remote and server/local addresses and ports to http/https request logging.
+# Defaults to true when either webServiceHaProxyProtocolEnabled or webServiceTrustXForwardedFor is enabled.
+webServiceLogDetailedAddresses=
+
 # Name of the pulsar cluster to connect to
 clusterName=
 

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",
@@ -1111,6 +1121,9 @@ public long getNumberOfEntriesInBacklog(boolean isPrecise) {
                     messagesConsumedCounter, markDeletePosition, readPosition);
         }
         if (isPrecise) {
+            if (markDeletePosition.compareTo(ledger.getLastPosition()) >= 0) {
+                return 0;
+            }
             return getNumberOfEntries(Range.openClosed(markDeletePosition, ledger.getLastPosition()));
         }
 
@@ -1265,12 +1278,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 +1550,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();