skip to get kafka advertised addres from zk when lookup with listenerName (#670)

This is the first step for #669, only support one listener which configed in `kafkaListenerName`, and only support PLAINTEXT

Co-authored-by: wangjialing <wangjialing@cmss.chinamobile.com>

Author: 2 people

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

@@ -156,6 +156,7 @@
 import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
 import org.apache.kafka.common.requests.WriteTxnMarkersRequest;
 import org.apache.kafka.common.requests.WriteTxnMarkersResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
@@ -177,6 +178,7 @@
 import org.apache.pulsar.metadata.api.MetadataCache;
 import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
 import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
+import org.eclipse.jetty.util.StringUtil;
 
 /**
  * This class contains all the request handling methods.
@@ -2108,6 +2110,22 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         if (future != null) {
             return future;
         }
+
+        // if kafkaListenerName is set, the lookup result is the advertised address
+        if (!StringUtil.isBlank(kafkaConfig.getKafkaListenerName())) {
+            // TODO:　should add SecurityProtocol according to which endpoint is handling the request.
+            //  firstly we only support PLAINTEXT when lookup with kafkaListenerName
+            String kafkaAdvertisedAddress = String.format("%s://%s:%s", SecurityProtocol.PLAINTEXT.name(),
+                    pulsarAddress.getHostName(), pulsarAddress.getPort());
+            KafkaTopicManager.KOP_ADDRESS_CACHE.put(topic.toString(), returnFuture);
+            returnFuture.complete(Optional.ofNullable(kafkaAdvertisedAddress));
+            if (log.isDebugEnabled()) {
+                log.debug("{} get kafka Advertised Address through kafkaListenerName: {}",
+                        topic, pulsarAddress);
+            }
+            return returnFuture;
+        }
+
         // advertised data is write in  /loadbalance/brokers/advertisedAddress:webServicePort
         // here we get the broker url, need to find related webServiceUrl.
         pulsarService.getPulsarResources()

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopBrokerLookupManager.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.loadbalance.LoadManager;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
@@ -32,6 +33,7 @@
 import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
 import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
 import org.apache.pulsar.zookeeper.ZooKeeperCache;
+import org.eclipse.jetty.util.StringUtil;
 
 
 /**
@@ -165,6 +167,22 @@ private CompletableFuture<Optional<String>> getProtocolDataToAdvertise(
         if (KOP_ADDRESS_CACHE.containsKey(topic.toString())) {
             return KOP_ADDRESS_CACHE.get(topic.toString());
         }
+
+        // if kafkaListenerName is set, the lookup result is the advertised address
+        if (!StringUtil.isBlank(lookupClient.getPulsarClient().getConfiguration().getListenerName())) {
+            // TODO:　should add SecurityProtocol according to which endpoint is handling the request.
+            //  firstly we only support PLAINTEXT when lookup with kafkaListenerName
+            String kafkaAdvertisedAddress = String.format("%s://%s:%s", SecurityProtocol.PLAINTEXT.name(),
+                    pulsarAddress.getHostName(), pulsarAddress.getPort());
+            KOP_ADDRESS_CACHE.put(topic.toString(), kopAddressFuture);
+            if (log.isDebugEnabled()) {
+                log.debug("{} get kafka Advertised Address through kafkaListenerName: {}",
+                        topic, pulsarAddress);
+            }
+            kopAddressFuture.complete(Optional.ofNullable(kafkaAdvertisedAddress));
+            return kopAddressFuture;
+        }
+
         // advertised data is write in  /loadbalance/brokers/advertisedAddress:webServicePort
         // here we get the broker url, need to find related webServiceUrl.
         ZooKeeperCache zkCache = pulsarService.getLocalZkCache();

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java

@@ -39,15 +39,13 @@ protected void cleanup() throws Exception {
     public void testListenerName() throws Exception {
         super.resetConfig();
         conf.setAdvertisedAddress(null);
-        conf.setInternalListenerName("external");
-        // There's a limit that PulsarService doesn't use advertised listener's address as it's brokerServiceUrl's
-        // address. So here the "external" listener's port should be the same with brokerPort and address should be
-        // localhost.
         final String localAddress = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(null);
+        conf.setInternalListenerName("pulsar");
         final String advertisedListeners =
-                "internal:pulsar://192.168.0.2:6650,external:pulsar://" + localAddress + ":" + brokerPort;
+                "pulsar:pulsar://" + localAddress + ":" + brokerPort
+                        + ",kafka:pulsar://" + "localhost:" + kafkaBrokerPort;
         conf.setAdvertisedListeners(advertisedListeners);
-        conf.setKafkaListenerName("external");
+        conf.setKafkaListenerName("kafka");
         log.info("Set advertisedListeners to {}", advertisedListeners);
         super.internalSetup();