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

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

@@ -499,6 +499,8 @@ The Apache Software License, Version 2.0
     - io.reactivex.rxjava3-rxjava-3.0.1.jar
   * RoaringBitmap
     - org.roaringbitmap-RoaringBitmap-1.2.0.jar
+  * Spotify completable-futures
+    - com.spotify-completable-futures-0.3.6.jar
 
 BSD 3-clause "New" or "Revised" License
  * Google auth library
@@ -543,16 +545,16 @@ CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt
     - org.glassfish.hk2-osgi-resource-locator-1.0.3.jar
     - org.glassfish.hk2.external-aopalliance-repackaged-2.6.1.jar
  * Jersey
-    - org.glassfish.jersey.containers-jersey-container-servlet-2.34.jar
-    - org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar
-    - org.glassfish.jersey.core-jersey-client-2.34.jar
-    - org.glassfish.jersey.core-jersey-common-2.34.jar
-    - org.glassfish.jersey.core-jersey-server-2.34.jar
-    - org.glassfish.jersey.ext-jersey-entity-filtering-2.34.jar
-    - org.glassfish.jersey.media-jersey-media-json-jackson-2.34.jar
-    - org.glassfish.jersey.media-jersey-media-multipart-2.34.jar
-    - org.glassfish.jersey.inject-jersey-hk2-2.34.jar
- * Mimepull -- org.jvnet.mimepull-mimepull-1.9.13.jar
+    - org.glassfish.jersey.containers-jersey-container-servlet-2.42.jar
+    - org.glassfish.jersey.containers-jersey-container-servlet-core-2.42.jar
+    - org.glassfish.jersey.core-jersey-client-2.42.jar
+    - org.glassfish.jersey.core-jersey-common-2.42.jar
+    - org.glassfish.jersey.core-jersey-server-2.42.jar
+    - org.glassfish.jersey.ext-jersey-entity-filtering-2.42.jar
+    - org.glassfish.jersey.media-jersey-media-json-jackson-2.42.jar
+    - org.glassfish.jersey.media-jersey-media-multipart-2.42.jar
+    - org.glassfish.jersey.inject-jersey-hk2-2.42.jar
+ * Mimepull -- org.jvnet.mimepull-mimepull-1.9.15.jar
 
 Eclipse Distribution License 1.0 -- ../licenses/LICENSE-EDL-1.0.txt
  * Jakarta Activation

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

@@ -409,6 +409,7 @@ The Apache Software License, Version 2.0
   * Apache Avro
     - avro-1.11.3.jar
     - avro-protobuf-1.11.3.jar
+ * Spotify completable-futures -- completable-futures-0.3.6.jar
 
 BSD 3-clause "New" or "Revised" License
  * JSR305 -- jsr305-3.0.2.jar -- ../licenses/LICENSE-JSR305.txt
@@ -438,13 +439,13 @@ CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt
     - aopalliance-repackaged-2.6.1.jar
     - osgi-resource-locator-1.0.3.jar
  * Jersey
-    - jersey-client-2.34.jar
-    - jersey-common-2.34.jar
-    - jersey-entity-filtering-2.34.jar
-    - jersey-media-json-jackson-2.34.jar
-    - jersey-media-multipart-2.34.jar
-    - jersey-hk2-2.34.jar
- * Mimepull -- mimepull-1.9.13.jar
+    - jersey-client-2.42.jar
+    - jersey-common-2.42.jar
+    - jersey-entity-filtering-2.42.jar
+    - jersey-media-json-jackson-2.42.jar
+    - jersey-media-multipart-2.42.jar
+    - jersey-hk2-2.42.jar
+ * Mimepull -- mimepull-1.9.15.jar
 
 Eclipse Distribution License 1.0 -- ../licenses/LICENSE-EDL-1.0.txt
  * Jakarta Activation

docker/pulsar/scripts/gen-yml-from-env.py

@@ -50,6 +50,9 @@
     'brokerClientTlsProtocols',
     'webServiceTlsCiphers',
     'webServiceTlsProtocols',
+    'additionalJavaRuntimeArguments',
+    'additionalEnabledConnectorUrlPatterns',
+    'additionalEnabledFunctionsUrlPatterns'
 ]
 
 PF_ENV_PREFIX = 'PF_'

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

@@ -420,7 +420,8 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
             }, null);
             return future;
         }).thenAccept(ml -> callback.openLedgerComplete(ml, ctx)).exceptionally(exception -> {
-            callback.openLedgerFailed((ManagedLedgerException) exception.getCause(), ctx);
+            callback.openLedgerFailed(ManagedLedgerException
+                    .getManagedLedgerException(FutureUtil.unwrapCompletionException(exception)), ctx);
             return null;
         });
     }
@@ -444,7 +445,8 @@ public void asyncOpenReadOnlyManagedLedger(String managedLedgerName,
 
         }).exceptionally(e -> {
             log.error("[{}] Failed to initialize Read-only managed ledger", managedLedgerName, e);
-            callback.openReadOnlyManagedLedgerFailed((ManagedLedgerException) e.getCause(), ctx);
+            callback.openReadOnlyManagedLedgerFailed(ManagedLedgerException
+                            .getManagedLedgerException(FutureUtil.unwrapCompletionException(e)), ctx);
             return null;
         });
     }

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

@@ -983,7 +983,8 @@ public synchronized void asyncOpenCursor(final String cursorName, final InitialP
         if (uninitializedCursors.containsKey(cursorName)) {
             uninitializedCursors.get(cursorName).thenAccept(cursor -> callback.openCursorComplete(cursor, ctx))
                     .exceptionally(ex -> {
-                callback.openCursorFailed((ManagedLedgerException) ex, ctx);
+                callback.openCursorFailed(ManagedLedgerException
+                        .getManagedLedgerException(FutureUtil.unwrapCompletionException(ex)), ctx);
                 return null;
             });
             return;
@@ -3014,9 +3015,8 @@ public void asyncDelete(final DeleteLedgerCallback callback, final Object ctx) {
         truncateFuture.whenComplete((ignore, exc) -> {
             if (exc != null) {
                 log.error("[{}] Error truncating ledger for deletion", name, exc);
-                callback.deleteLedgerFailed(exc instanceof ManagedLedgerException
-                        ? (ManagedLedgerException) exc : new ManagedLedgerException(exc),
-                        ctx);
+                callback.deleteLedgerFailed(ManagedLedgerException.getManagedLedgerException(
+                        FutureUtil.unwrapCompletionException(exc)), ctx);
             } else {
                 asyncDeleteInternal(callback, ctx);
             }

pom.xml

@@ -147,7 +147,7 @@ flexible messaging model and an intuitive client API.</description>
     <netty-iouring.version>0.0.21.Final</netty-iouring.version>
     <jetty.version>9.4.54.v20240208</jetty.version>
     <conscrypt.version>2.5.2</conscrypt.version>
-    <jersey.version>2.34</jersey.version>
+    <jersey.version>2.42</jersey.version>
     <athenz.version>1.10.50</athenz.version>
     <prometheus.version>0.16.0</prometheus.version>
     <vertx.version>4.5.8</vertx.version>
@@ -255,6 +255,7 @@ flexible messaging model and an intuitive client API.</description>
     <disruptor.version>3.4.3</disruptor.version>
     <zstd-jni.version>1.5.2-3</zstd-jni.version>
     <netty-reactive-streams.version>2.0.6</netty-reactive-streams.version>
+    <completable-futures.version>0.3.6</completable-futures.version>
     <failsafe.version>3.3.2</failsafe.version>
 
     <!-- test dependencies -->
@@ -646,6 +647,12 @@ flexible messaging model and an intuitive client API.</description>
         <version>${bookkeeper.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>com.spotify</groupId>
+        <artifactId>completable-futures</artifactId>
+        <version>${completable-futures.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>org.rocksdb</groupId>
         <artifactId>rocksdbjni</artifactId>

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderList.java

@@ -120,7 +120,8 @@ public CompletableFuture<AuthData> authenticateAsync(AuthData authData) {
                             if (log.isDebugEnabled()) {
                                 log.debug("Authentication failed for auth provider " + authState.getClass() + ": ", ex);
                             }
-                            authenticateRemainingAuthStates(authChallengeFuture, authData, ex, states.size() - 1);
+                            authenticateRemainingAuthStates(authChallengeFuture, authData, ex,
+                                    states.isEmpty() ? -1 : 0);
                         }
                     });
             return authChallengeFuture;
@@ -130,7 +131,7 @@ private void authenticateRemainingAuthStates(CompletableFuture<AuthData> authCha
                                                      AuthData clientAuthData,
                                                      Throwable previousException,
                                                      int index) {
-            if (index < 0) {
+            if (index < 0 || index >= states.size()) {
                 if (previousException == null) {
                     previousException = new AuthenticationException("Authentication required");
                 }
@@ -142,7 +143,7 @@ private void authenticateRemainingAuthStates(CompletableFuture<AuthData> authCha
             AuthenticationState state = states.get(index);
             if (state == authState) {
                 // Skip the current auth state
-                authenticateRemainingAuthStates(authChallengeFuture, clientAuthData, null, index - 1);
+                authenticateRemainingAuthStates(authChallengeFuture, clientAuthData, null, index + 1);
             } else {
                 state.authenticateAsync(clientAuthData)
                         .whenComplete((authChallenge, ex) -> {
@@ -155,7 +156,7 @@ private void authenticateRemainingAuthStates(CompletableFuture<AuthData> authCha
                                     log.debug("Authentication failed for auth provider "
                                             + authState.getClass() + ": ", ex);
                                 }
-                                authenticateRemainingAuthStates(authChallengeFuture, clientAuthData, ex, index - 1);
+                                authenticateRemainingAuthStates(authChallengeFuture, clientAuthData, ex, index + 1);
                             }
                         });
             }
@@ -228,15 +229,15 @@ public String getAuthMethodName() {
     @Override
     public CompletableFuture<String> authenticateAsync(AuthenticationDataSource authData) {
         CompletableFuture<String> roleFuture = new CompletableFuture<>();
-        authenticateRemainingAuthProviders(roleFuture, authData, null, providers.size() - 1);
+        authenticateRemainingAuthProviders(roleFuture, authData, null, providers.isEmpty() ? -1 : 0);
         return roleFuture;
     }
 
     private void authenticateRemainingAuthProviders(CompletableFuture<String> roleFuture,
                                                     AuthenticationDataSource authData,
                                                     Throwable previousException,
                                                     int index) {
-        if (index < 0) {
+        if (index < 0 || index >= providers.size()) {
             if (previousException == null) {
                 previousException = new AuthenticationException("Authentication required");
             }
@@ -254,7 +255,7 @@ private void authenticateRemainingAuthProviders(CompletableFuture<String> roleFu
                         if (log.isDebugEnabled()) {
                             log.debug("Authentication failed for auth provider " + provider.getClass() + ": ", ex);
                         }
-                        authenticateRemainingAuthProviders(roleFuture, authData, ex, index - 1);
+                        authenticateRemainingAuthProviders(roleFuture, authData, ex, index + 1);
                     }
                 });
         }

pulsar-broker/src/main/java/org/apache/pulsar/PulsarClusterMetadataSetup.java

@@ -94,6 +94,19 @@ private static class Arguments {
                 description = "Broker-service URL for new cluster with TLS encryption", required = false)
         private String clusterBrokerServiceUrlTls;
 
+        @Parameter(names = {"-te",
+                "--tls-enable"},
+                description = "Enable TLS connection for new cluster")
+        private Boolean clusterBrokerClientTlsEnabled;
+
+        @Parameter(names = "--auth-plugin",
+                description = "The authentication plugin for new cluster")
+        protected String clusterAuthenticationPlugin;
+
+        @Parameter(names = "--auth-parameters",
+                description = "The authentication parameters for new cluster")
+        protected String clusterAuthenticationParameters;
+
         @Parameter(names = { "-zk",
                 "--zookeeper" }, description = "Local ZooKeeper quorum connection string",
                 required = false,
@@ -315,14 +328,36 @@ private static void initializeCluster(Arguments arguments, int bundleNumberForDe
 
         PulsarResources resources = new PulsarResources(localStore, configStore);
 
-        ClusterData clusterData = ClusterData.builder()
-                .serviceUrl(arguments.clusterWebServiceUrl)
-                .serviceUrlTls(arguments.clusterWebServiceUrlTls)
-                .brokerServiceUrl(arguments.clusterBrokerServiceUrl)
-                .brokerServiceUrlTls(arguments.clusterBrokerServiceUrlTls)
-                .proxyServiceUrl(arguments.clusterProxyUrl)
-                .proxyProtocol(arguments.clusterProxyProtocol)
-                .build();
+        ClusterData.Builder clusterDataBuilder = ClusterData.builder();
+        if (arguments.clusterWebServiceUrl != null) {
+            clusterDataBuilder.serviceUrl(arguments.clusterWebServiceUrl);
+        }
+        if (arguments.clusterWebServiceUrlTls != null) {
+            clusterDataBuilder.serviceUrlTls(arguments.clusterWebServiceUrlTls);
+        }
+        if (arguments.clusterBrokerServiceUrl != null) {
+            clusterDataBuilder.brokerServiceUrl(arguments.clusterBrokerServiceUrl);
+        }
+        if (arguments.clusterBrokerServiceUrlTls != null) {
+            clusterDataBuilder.brokerServiceUrlTls(arguments.clusterBrokerServiceUrlTls);
+        }
+        if (arguments.clusterBrokerClientTlsEnabled != null) {
+            clusterDataBuilder.brokerClientTlsEnabled(arguments.clusterBrokerClientTlsEnabled);
+        }
+        if (arguments.clusterAuthenticationPlugin != null) {
+            clusterDataBuilder.authenticationPlugin(arguments.clusterAuthenticationPlugin);
+        }
+        if (arguments.clusterAuthenticationParameters != null) {
+            clusterDataBuilder.authenticationParameters(arguments.clusterAuthenticationParameters);
+        }
+        if (arguments.clusterProxyUrl != null) {
+            clusterDataBuilder.proxyServiceUrl(arguments.clusterProxyUrl);
+        }
+        if (arguments.clusterProxyProtocol != null) {
+            clusterDataBuilder.proxyProtocol(arguments.clusterProxyProtocol);
+        }
+
+        ClusterData clusterData = clusterDataBuilder.build();
         if (!resources.getClusterResources().clusterExists(arguments.cluster)) {
             resources.getClusterResources().createCluster(arguments.cluster, clusterData);
         }

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -882,7 +882,7 @@ public void start() throws PulsarServerException {
                 MLTransactionMetadataStoreProvider.initBufferedWriterMetrics(getAdvertisedAddress());
                 MLPendingAckStoreProvider.initBufferedWriterMetrics(getAdvertisedAddress());
 
-                this.transactionBufferSnapshotServiceFactory = new TransactionBufferSnapshotServiceFactory(getClient());
+                this.transactionBufferSnapshotServiceFactory = new TransactionBufferSnapshotServiceFactory(this);
 
                 this.transactionTimer =
                         new HashedWheelTimer(new DefaultThreadFactory("pulsar-transaction-timer"));