Compatible mater code (#1093)

Author: coderzc

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

@@ -14,7 +14,9 @@
 package io.streamnative.pulsar.handlers.amqp;
 
 import io.netty.channel.ChannelHandlerContext;
+import java.util.Optional;
 import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
 import org.apache.pulsar.broker.service.Consumer;
 import org.apache.pulsar.broker.service.ServerCnx;
 
@@ -30,7 +32,7 @@ public AmqpPulsarServerCnx(PulsarService pulsar, ChannelHandlerContext ctx) {
     }
 
     @Override
-    public void closeConsumer(Consumer consumer) {
+    public void closeConsumer(Consumer consumer, Optional<BrokerLookupData> assignedBrokerLookupData) {
         // avoid close the connection when closing the consumer
     }
 }

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/proxy/PulsarServiceLookupHandler.java

@@ -16,7 +16,6 @@
 import io.streamnative.pulsar.handlers.amqp.AmqpProtocolHandler;
 import java.io.Closeable;
 import java.io.IOException;
-import java.net.InetSocketAddress;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.List;
@@ -27,6 +26,7 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.resources.MetadataStoreCacheLoader;
+import org.apache.pulsar.client.impl.LookupTopicResult;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
@@ -53,19 +53,19 @@ public CompletableFuture<Pair<String, Integer>> findBroker(TopicName topicName,
         CompletableFuture<Pair<String, Integer>> lookupResult = new CompletableFuture<>();
 
         // lookup the broker for the given topic
-        CompletableFuture<Pair<InetSocketAddress, InetSocketAddress>> lookup =
+        CompletableFuture<LookupTopicResult> lookup =
                 pulsarClient.getLookup().getBroker(topicName);
         lookup.whenComplete((result, throwable) -> {
             if (throwable != null) {
                 lookupResult.completeExceptionally(throwable);
                 return;
             }
-            if (result == null || result.getLeft() == null) {
+            if (result == null || result.getLogicalAddress() == null) {
                 lookupResult.completeExceptionally(new ProxyException(
                         "Unable to resolve the broker for the topic: " + topicName));
                 return;
             }
-            String hostAndPort = result.getLeft().getHostName() + ":" + result.getLeft().getPort();
+            String hostAndPort = result.getLogicalAddress().getHostName() + ":" + result.getLogicalAddress().getPort();
 
             // fetch the protocol handler data
             List<LoadManagerReport> brokers = metadataStoreCacheLoader.getAvailableBrokers();

pom.xml

@@ -40,7 +40,7 @@
     <project.compiler.release>${maven.compiler.target}</project.compiler.release>
 
     <!-- dependencies -->
-    <pulsar.version>3.0.0.1-SNAPSHOT</pulsar.version>
+    <pulsar.version>3.2.0-SNAPSHOT</pulsar.version>
     <qpid-protocol-plugin.version>8.0.0</qpid-protocol-plugin.version>
     <rabbitmq.version>5.8.0</rabbitmq.version>