Bump Pulsar to 5.0.0-M1-SNAPSHOT (#1848)

Author: gaoran10

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/admin/impl/BaseResources.java

@@ -20,7 +20,6 @@
 import io.streamnative.pulsar.handlers.amqp.QueueService;
 import io.streamnative.pulsar.handlers.amqp.admin.model.VhostBean;
 import java.net.URI;
-import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -143,40 +142,36 @@ private PulsarService pulsar() {
         return aop().getBrokerService().getPulsar();
     }
 
-    public boolean isRequestHttps() {
-        return "https".equalsIgnoreCase(httpRequest.getScheme());
-    }
-
     protected CompletableFuture<Void> validateTopicOwnershipAsync(TopicName topicName, boolean authoritative) {
         NamespaceService nsService = pulsar().getNamespaceService();
 
         LookupOptions options = LookupOptions.builder()
                 .authoritative(authoritative)
-                .requestHttps(isRequestHttps())
                 .readOnly(false)
                 .loadTopicsInBundle(false)
                 .build();
 
-        return nsService.getWebServiceUrlAsync(topicName, options)
-                .thenApply(webUrl -> {
+        return nsService.getLookupResultForWebRequestAsync(topicName, options)
+                .thenApply(lookupResult -> {
                     // Ensure we get a url
-                    if (webUrl == null || !webUrl.isPresent()) {
-                        log.info("Unable to get web service url");
+                    if (lookupResult.isEmpty()) {
+                        log.warn("Unable to get lookup result for topic: {}, authoritative: {}",
+                                topicName, authoritative);
                         throw new RestException(Response.Status.PRECONDITION_FAILED,
                                 "Failed to find ownership for topic:" + topicName);
                     }
-                    return webUrl.get();
+                    return lookupResult.get();
                 }).thenCompose(webUrl -> nsService.isServiceUnitOwnedAsync(topicName)
                         .thenApply(isTopicOwned -> Pair.of(webUrl, isTopicOwned))
                 ).thenAccept(pair -> {
-                    URL webUrl = pair.getLeft();
+                    URI webUri = pair.getLeft().toLookupRedirectUri(uri.getRequestUri());
                     boolean isTopicOwned = pair.getRight();
 
                     if (!isTopicOwned) {
                         boolean newAuthoritative = isLeaderBroker(pulsar());
                         // Replace the host and port of the current request and redirect
                         URI redirect = UriBuilder.fromUri(uri.getRequestUri())
-                                .host(webUrl.getHost())
+                                .host(webUri.getHost())
                                 .port(aop().getAmqpConfig().getAmqpAdminPort())
                                 .replaceQueryParam("authoritative", newAuthoritative)
                                 .build();

pom.xml

@@ -40,10 +40,10 @@
     <project.compiler.release>${maven.compiler.target}</project.compiler.release>
 
     <!-- dependencies -->
-    <pulsar.version>4.2.0-SNAPSHOT</pulsar.version>
+    <pulsar.version>5.0.0-M1-SNAPSHOT</pulsar.version>
+    <sn.bom.version>5.0.0-M1-SNAPSHOT</sn.bom.version>
     <qpid-protocol-plugin.version>8.0.0</qpid-protocol-plugin.version>
     <rabbitmq.version>5.8.0</rabbitmq.version>
-    <sn.bom.version>4.2.0-SNAPSHOT</sn.bom.version>
     <opentelemetry.version>1.56.0</opentelemetry.version>
 
     <!-- test dependencies -->