Support multiple listeners without configuring broker's advertisedListeners (#864)

Fixes #818

### Motivation

#742 adds the multiple listeners support for KoP 2.8.x. However, there're some flaws. 

First, the KoP advertised listeners must be added to `advertisedListeners` config. This config should be treated as the advertised listeners for broker, not protocol handlers. What `KafkaProtocolHandler#getProtocolDataToAdvertise` returns is not used though it has been written to ZK. It also makes `kafkaAdvertisedListeners` meaningless and it was marked as deprecated.

### Modification

Benefit by [PIP 95](apache/pulsar#12040), Pulsar client doesn't need to configure listener name for topic lookup. So this PR simplifies the implementation of `LookupClient` by using a shared `PulsarClient` without configuring listener name.

Then this PR uses the `MetadataStoreCacheLoader` introduced from #820 to get the `kafkaAdvertisedListeners` from ZK. Since it has already been cached, this PR removes the `KOP_ADDRESS_CACHE`.

To verify the feature, this PR adds a test `testMetadataRequestForMultiListeners`. The KoP config is like

```properties
kafkaListeners=PLAINTEXT://0.0.0.0:<port1>,GW://0.0.0.0:<port2>
kafkaProtocolMap=PLAINTEXT:PLAINTEXT,GW:PLAINTEXT
kafkaAdvertisedListeners=PLAINTEXT://192.168.0.1:<port3>,GW://192.168.0.1:<port4>
```

And verify that each `KafkaRequestHandler` can handle the METADATA requests well.

Author: BewareMyPower

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

@@ -90,6 +90,18 @@ public InetSocketAddress getInetAddress() {
         return new InetSocketAddress(hostname, port);
     }
 
+    // listeners must be enable to be split into at least 1 token
+    private static String[] getListenerArray(final String listeners) {
+        if (StringUtils.isEmpty(listeners)) {
+            throw new IllegalStateException("listeners is empty");
+        }
+        final String[] listenerArray = listeners.split(END_POINT_SEPARATOR);
+        if (listenerArray.length == 0) {
+            throw new IllegalStateException(listeners + " is split into 0 tokens by " + END_POINT_SEPARATOR);
+        }
+        return listenerArray;
+    }
+
     @VisibleForTesting
     public static Map<String, EndPoint> parseListeners(final String listeners) {
         return parseListeners(listeners, "");
@@ -98,7 +110,7 @@ public static Map<String, EndPoint> parseListeners(final String listeners) {
     private static Map<String, EndPoint> parseListeners(final String listeners,
                                                         final Map<String, SecurityProtocol> protocolMap) {
         final Map<String, EndPoint> endPointMap = new HashMap<>();
-        for (String listener : listeners.split(END_POINT_SEPARATOR)) {
+        for (String listener : getListenerArray(listeners)) {
             final EndPoint endPoint = new EndPoint(listener, protocolMap);
             if (endPointMap.containsKey(endPoint.listenerName)) {
                 throw new IllegalStateException(
@@ -115,6 +127,22 @@ public static Map<String, EndPoint> parseListeners(final String listeners, final
         return parseListeners(listeners, parseProtocolMap(protocolMapString));
     }
 
+    public static String findListener(final String listeners, final String name) {
+        if (name == null) {
+            return null;
+        }
+        for (String listener : getListenerArray(listeners)) {
+            if (listener.contains(":") && listener.substring(0, listener.indexOf(":")).equals(name)) {
+                return listener;
+            }
+        }
+        throw new IllegalStateException("listener \"" + name + "\" doesn't exist in " + listeners);
+    }
+
+    public static String findFirstListener(String listeners) {
+        return getListenerArray(listeners)[0];
+    }
+
     public static EndPoint getPlainTextEndPoint(final String listeners) {
         for (String listener : listeners.split(END_POINT_SEPARATOR)) {
             if (listener.startsWith(SecurityProtocol.PLAINTEXT.name())

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

@@ -15,6 +15,7 @@
 
 import static io.streamnative.pulsar.handlers.kop.KafkaProtocolHandler.TLS_HANDLER;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
@@ -92,10 +93,13 @@ protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast(new LengthFieldPrepender(4));
         ch.pipeline().addLast("frameDecoder",
             new LengthFieldBasedFrameDecoder(MAX_FRAME_LENGTH, 0, 4, 0, 4));
-        ch.pipeline().addLast("handler",
-                new KafkaRequestHandler(pulsarService, kafkaConfig,
-                        tenantContextManager, kopBrokerLookupManager, adminManager,
-                        enableTls, advertisedEndPoint, statsLogger));
+        ch.pipeline().addLast("handler", newCnx());
     }
 
+    @VisibleForTesting
+    public KafkaRequestHandler newCnx() throws Exception {
+        return new KafkaRequestHandler(pulsarService, kafkaConfig,
+                tenantContextManager, kopBrokerLookupManager, adminManager,
+                enableTls, advertisedEndPoint, statsLogger);
+    }
 }

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

@@ -85,6 +85,9 @@ public class KafkaProtocolHandler implements ProtocolHandler, TenantContextManag
     private KopBrokerLookupManager kopBrokerLookupManager;
     private AdminManager adminManager = null;
     private SystemTopicClient txnTopicClient;
+    @VisibleForTesting
+    @Getter
+    private Map<InetSocketAddress, ChannelInitializer<SocketChannel>> channelInitializerMap;
 
     @Getter
     @VisibleForTesting
@@ -577,7 +580,8 @@ public Map<InetSocketAddress, ChannelInitializer<SocketChannel>> newChannelIniti
                     forEach((listener, endPoint) ->
                             builder.put(endPoint.getInetAddress(), newKafkaChannelInitializer(endPoint))
                     );
-            return builder.build();
+            channelInitializerMap = builder.build();
+            return channelInitializerMap;
         } catch (Exception e){
             log.error("KafkaProtocolHandler newChannelInitializers failed with ", e);
             return null;

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

@@ -203,7 +203,6 @@ public class KafkaRequestHandler extends KafkaCommandDecoder {
 
     private final Boolean tlsEnabled;
     private final EndPoint advertisedEndPoint;
-    private final String advertisedListeners;
     private final int defaultNumPartitions;
     public final int maxReadEntriesNum;
     private final int failedAuthenticationDelayMs;
@@ -308,7 +307,6 @@ public KafkaRequestHandler(PulsarService pulsarService,
         this.adminManager = adminManager;
         this.tlsEnabled = tlsEnabled;
         this.advertisedEndPoint = advertisedEndPoint;
-        this.advertisedListeners = kafkaConfig.getKafkaAdvertisedListeners();
         this.topicManager = new KafkaTopicManager(this);
         this.defaultNumPartitions = kafkaConfig.getDefaultNumPartitions();
         this.maxReadEntriesNum = kafkaConfig.getMaxReadEntriesNum();
@@ -2481,7 +2479,7 @@ public CompletableFuture<PartitionMetadata> findBroker(TopicName topic) {
         if (log.isDebugEnabled()) {
             log.debug("[{}] Handle Lookup for {}", ctx.channel(), topic);
         }
-        return kopBrokerLookupManager.findBroker(topic, advertisedEndPoint)
+        return kopBrokerLookupManager.findBroker(topic.toString(), advertisedEndPoint)
                 .thenApply(listenerInetSocketAddressOpt -> listenerInetSocketAddressOpt
                         .map(inetSocketAddress -> newPartitionMetadata(topic, newNode(inetSocketAddress)))
                         .orElse(null)

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

@@ -184,6 +184,8 @@ public class KafkaServiceConfiguration extends ServiceConfiguration {
         category = CATEGORY_KOP,
         doc = "Comma-separated list of URIs we will listen on and the listener names.\n"
                 + "e.g. PLAINTEXT://localhost:9092,SSL://localhost:9093.\n"
+                + "Each URI's scheme represents a listener name if `kafkaProtocolMap` is configured.\n"
+                + "Otherwise, the scheme must be a valid protocol in [PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL].\n"
                 + "If hostname is not set, bind to the default interface."
     )
     private String kafkaListeners;
@@ -195,10 +197,10 @@ public class KafkaServiceConfiguration extends ServiceConfiguration {
     )
     private String kafkaProtocolMap;
 
-    @Deprecated
     @FieldContext(
             category = CATEGORY_KOP,
-            doc = "Use kafkaProtocolMap, kafkaListeners and advertisedAddress instead."
+            doc = "Listeners to publish to ZooKeeper for clients to use.\n"
+                    + "The format is the same as `kafkaListeners`.\n"
     )
     private String kafkaAdvertisedListeners;
 

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

@@ -21,8 +21,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
 import javax.annotation.Nullable;
-import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.PulsarService;
@@ -38,149 +38,79 @@
 @Slf4j
 public class KopBrokerLookupManager {
 
-    private final String advertisedListeners;
     private final LookupClient lookupClient;
     private final MetadataStoreCacheLoader metadataStoreCacheLoader;
 
     private final AtomicBoolean closed = new AtomicBoolean(false);
 
-    public static final ConcurrentHashMap<String, ConcurrentHashMap<String, CompletableFuture<InetSocketAddress>>>
+    public static final ConcurrentHashMap<String, CompletableFuture<InetSocketAddress>>
             LOOKUP_CACHE = new ConcurrentHashMap<>();
 
-    public static final ConcurrentHashMap<String, CompletableFuture<Optional<String>>>
-            KOP_ADDRESS_CACHE = new ConcurrentHashMap<>();
-
     public KopBrokerLookupManager(KafkaServiceConfiguration conf, PulsarService pulsarService) throws Exception {
-        this.advertisedListeners = conf.getKafkaAdvertisedListeners();
         this.lookupClient = KafkaProtocolHandler.getLookupClient(pulsarService);
         this.metadataStoreCacheLoader = new MetadataStoreCacheLoader(pulsarService.getPulsarResources(),
                 conf.getBrokerLookupTimeoutMs());
     }
 
-    public CompletableFuture<Optional<InetSocketAddress>> findBroker(@NonNull TopicName topic,
+    public CompletableFuture<Optional<InetSocketAddress>> findBroker(String topic,
                                                                      @Nullable EndPoint advertisedEndPoint) {
-        if (log.isDebugEnabled()) {
-            log.debug("Handle Lookup for topic {}", topic);
-        }
-        CompletableFuture<Optional<InetSocketAddress>> returnFuture = new CompletableFuture<>();
-
-        getTopicBroker(topic.toString(),
-                advertisedEndPoint != null && advertisedEndPoint.isValidInProtocolMap()
-                        ? advertisedEndPoint.getListenerName() : null)
-                .thenApply(address -> getProtocolDataToAdvertise(address, topic, advertisedEndPoint))
-                .thenAccept(kopAddressFuture -> kopAddressFuture.thenAccept(listenersOptional -> {
-                    if (!listenersOptional.isPresent()) {
-                        log.error("Not get advertise data for Kafka topic:{}.", topic);
-                        removeTopicManagerCache(topic.toString());
-                        returnFuture.complete(Optional.empty());
-                        return;
+        return getTopicBroker(topic)
+                .thenApply(internalListenerAddress -> {
+                    if (internalListenerAddress == null) {
+                        log.error("[{}] failed get pulsar address, returned null.", topic);
+                        removeTopicManagerCache(topic);
+                        return Optional.empty();
+                    } else if (log.isDebugEnabled()) {
+                        log.debug("[{}] Found broker's internal listener address: {}",
+                                topic, internalListenerAddress);
                     }
 
-                    // It's the `kafkaAdvertisedListeners` config that's written to ZK
-                    final String listeners = listenersOptional.get();
-                    final EndPoint endPoint =
-                            (advertisedEndPoint != null && advertisedEndPoint.isTlsEnabled()
-                                    ? EndPoint.getSslEndPoint(listeners) : EndPoint.getPlainTextEndPoint(listeners));
-
-                    if (log.isDebugEnabled()) {
-                        log.debug("Found broker localListeners: {} for topicName: {}, "
-                                        + "localListeners: {}, found Listeners: {}",
-                                listeners, topic, advertisedListeners, listeners);
+                    try {
+                        final String listener = getAdvertisedListener(
+                                internalListenerAddress, topic, advertisedEndPoint);
+                        if (log.isDebugEnabled()) {
+                            log.debug("Found listener {} for topic {}", listener, topic);
+                        }
+                        final Matcher matcher = EndPoint.matcherListener(listener,
+                                listener + " cannot be split into 3 parts");
+                        return Optional.of(new InetSocketAddress(matcher.group(2), Integer.parseInt(matcher.group(3))));
+                    } catch (IllegalStateException | NumberFormatException e) {
+                        log.error("Failed to find the advertised listener: {}", e.getMessage());
+                        removeTopicManagerCache(topic);
+                        return Optional.empty();
                     }
-
-                    // here we found topic broker: broker2, but this is in broker1,
-                    // how to clean the lookup cache?
-                    if (!advertisedListeners.contains(endPoint.getOriginalListener())) {
-                        removeTopicManagerCache(topic.toString());
-                    }
-                    returnFuture.complete(Optional.of(endPoint.getInetAddress()));
-                })).exceptionally(throwable -> {
-                    log.error("Not get advertise data for Kafka topic:{}. throwable: [{}]",
-                            topic, throwable.getMessage());
-                    removeTopicManagerCache(topic.toString());
-                    returnFuture.complete(Optional.empty());
-                    return null;
                 });
-        return returnFuture;
     }
 
-    // call pulsarclient.lookup.getbroker to get and own a topic.
-    // when error happens, the returned future will complete with null.
-    public CompletableFuture<InetSocketAddress> getTopicBroker(String topicName, String listenerName) {
+    public CompletableFuture<InetSocketAddress> getTopicBroker(String topicName) {
         if (closed.get()) {
             if (log.isDebugEnabled()) {
                 log.debug("Return null for getTopicBroker({}) since channel closing", topicName);
             }
             return CompletableFuture.completedFuture(null);
         }
 
-        ConcurrentHashMap<String, CompletableFuture<InetSocketAddress>> topicLookupCache =
-                LOOKUP_CACHE.computeIfAbsent(topicName, t-> {
-                    if (log.isDebugEnabled()) {
-                        log.debug("Topic {} not in Lookup_cache, call lookupBroker", topicName);
-                    }
-                    ConcurrentHashMap<String, CompletableFuture<InetSocketAddress>> cache = new ConcurrentHashMap<>();
-                    cache.put(listenerName == null ? "" : listenerName, lookupBroker(topicName, listenerName));
-                    return cache;
-                });
-
-        return topicLookupCache.computeIfAbsent(listenerName == null ? "" : listenerName, t-> {
-            if (log.isDebugEnabled()) {
-                log.debug("Topic {} not in Lookup_cache, call lookupBroker", topicName);
-            }
-            return lookupBroker(topicName, listenerName);
-        });
+        if (log.isDebugEnabled()) {
+            log.debug("Handle Lookup for topic {}", topicName);
+        }
+        return LOOKUP_CACHE.computeIfAbsent(topicName, this::lookupBroker);
     }
 
-    private CompletableFuture<InetSocketAddress> lookupBroker(final String topic, String listenerName) {
+    private CompletableFuture<InetSocketAddress> lookupBroker(final String topic) {
         if (closed.get()) {
             if (log.isDebugEnabled()) {
                 log.debug("Return null for getTopic({}) since channel closing", topic);
             }
             return CompletableFuture.completedFuture(null);
         }
-        return lookupClient.getBrokerAddress(TopicName.get(topic), listenerName);
+        return lookupClient.getBrokerAddress(TopicName.get(topic));
     }
 
-    private CompletableFuture<Optional<String>> getProtocolDataToAdvertise(
-            InetSocketAddress pulsarAddress, TopicName topic, @Nullable EndPoint advertisedEndPoint) {
-        CompletableFuture<Optional<String>> returnFuture = new CompletableFuture<>();
-
-        if (pulsarAddress == null) {
-            log.error("[{}] failed get pulsar address, returned null.", topic.toString());
-
-            // getTopicBroker returns null. topic should be removed from LookupCache.
-            removeTopicManagerCache(topic.toString());
-
-            returnFuture.complete(Optional.empty());
-            return returnFuture;
-        }
+    private String getAdvertisedListener(InetSocketAddress internalListenerAddress,
+                                         String topic,
+                                         @Nullable EndPoint advertisedEndPoint) {
 
-        if (log.isDebugEnabled()) {
-            log.debug("Found broker for topic {} puslarAddress: {}",
-                    topic, pulsarAddress);
-        }
-
-        // get kop address from cache to prevent query zk each time.
-        final CompletableFuture<Optional<String>> future = KOP_ADDRESS_CACHE.get(topic.toString());
-        if (future != null) {
-            return future;
-        }
-
-        if (advertisedEndPoint != null && advertisedEndPoint.isValidInProtocolMap()) {
-            // if kafkaProtocolMap is set, the lookup result is the advertised address
-            String kafkaAdvertisedAddress = String.format("%s://%s:%s", advertisedEndPoint.getSecurityProtocol().name,
-                    pulsarAddress.getHostName(), pulsarAddress.getPort());
-            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;
-        }
-
-        List<LoadManagerReport> availableBrokers = metadataStoreCacheLoader.getAvailableBrokers();
+        final List<LoadManagerReport> availableBrokers = metadataStoreCacheLoader.getAvailableBrokers();
         if (log.isDebugEnabled()) {
             availableBrokers.forEach(loadManagerReport ->
                     log.debug("Handle getProtocolDataToAdvertise for {}, pulsarUrl: {}, "
@@ -193,18 +123,20 @@ private CompletableFuture<Optional<String>> getProtocolDataToAdvertise(
                             loadManagerReport.getProtocol(KafkaProtocolHandler.PROTOCOL_NAME)));
         }
 
-        String hostAndPort = pulsarAddress.getHostName() + ":" + pulsarAddress.getPort();
-        Optional<LoadManagerReport> serviceLookupData = availableBrokers.stream()
+        final String hostAndPort = internalListenerAddress.getHostName() + ":" + internalListenerAddress.getPort();
+        final Optional<LoadManagerReport> serviceLookupData = availableBrokers.stream()
                 .filter(loadManagerReport -> lookupDataContainsAddress(loadManagerReport, hostAndPort)).findAny();
-        if (serviceLookupData.isPresent()) {
-            KOP_ADDRESS_CACHE.put(topic.toString(), returnFuture);
-            returnFuture.complete(serviceLookupData.get().getProtocol(KafkaProtocolHandler.PROTOCOL_NAME));
-        } else {
-            log.error("No node for broker {} under loadBalance", pulsarAddress);
-            removeTopicManagerCache(topic.toString());
-            returnFuture.complete(Optional.empty());
+        if (!serviceLookupData.isPresent()) {
+            log.error("No node for broker {} under loadBalance", internalListenerAddress);
+            return null;
         }
-        return returnFuture;
+
+        return serviceLookupData.get().getProtocol(KafkaProtocolHandler.PROTOCOL_NAME).map(kafkaAdvertisedListeners ->
+            Optional.ofNullable(advertisedEndPoint)
+                    .map(endPoint -> EndPoint.findListener(kafkaAdvertisedListeners, endPoint.getListenerName()))
+                    .orElse(EndPoint.findFirstListener(kafkaAdvertisedListeners))
+        ).orElseThrow(() -> new IllegalStateException(
+                "No kafkaAdvertisedListeners found in broker " + internalListenerAddress));
     }
 
     // whether a ServiceLookupData contains wanted address.
@@ -215,12 +147,10 @@ private static boolean lookupDataContainsAddress(ServiceLookupData data, String
 
     public static void removeTopicManagerCache(String topicName) {
         LOOKUP_CACHE.remove(topicName);
-        KOP_ADDRESS_CACHE.remove(topicName);
     }
 
     public static void clear() {
         LOOKUP_CACHE.clear();
-        KOP_ADDRESS_CACHE.clear();
     }
 
     public void close() {