Update docs for multiple listeners and non-partitioned topic support

docs/configuration.md

@@ -5,14 +5,34 @@
 | Name                     | Description                                                  |
 | ------------------------ | ------------------------------------------------------------ |
 | kafkaListeners           | Comma-separated list of URIs that we will listen on and the listener names.<br>e.g. PLAINTEXT://localhost:9092,SSL://localhost:9093.<br>If the hostname is not set, the default interface is used. |
+| kafkaProtocolMap         | Comma-separated map of listener name and protocol.<br>e.g. PRIVATE:PLAINTEXT,PRIVATE_SSL:SSL,PUBLIC:PLAINTEXT,PUBLIC_SSL:SSL. |
 | listeners                | Deprecated. `kafkaListeners` is used.                   |
-| kafkaAdvertisedListeners | Listeners published to the ZooKeeper for clients to use.<br>The format is the same as `kafkaListeners`. |
-| kafkaListenerName        | Specify the internal listener name for the broker.<br>The listener name must be contained in the advertisedListeners.<br>This config is used as the listener name in topic lookup. |
+| kafkaAdvertisedListeners | Deprecated. Use kafkaProtocolMap, kafkaListeners and advertisedAddress instead. |
 
 > **NOTE**
 > 
 > Among all configurations, only `kafkaListeners` or `listeners` (deprecated) is required.
 
+To support multiple listeners, you need to specify different listener names in [`advertisedListeners`](https://pulsar.apache.org/docs/en/concepts-multiple-advertised-listeners/#use-multiple-advertised-listeners). Then map the listener name to the proper protocol in `kafkaProtocolMap`.
+
+For example, assuming you need to listen on port 9092 and 19092 with the `PLAINTEXT` protocol, the associated names are `kafka_internal` and `kafka_external`. Then you need to add the following configurations:
+
+```properties
+kafkaListeners=kafka_internal://localhost:9092,kafka_external://localhost:19092
+kafkaProtocolMap=kafka_internal:PLAINTEXT,kafka_external:PLAINTEXT
+advertisedListeners=pulsar:pulsar://localhost:6650,kafka_internal:pulsar://localhost:9092,kafka_external:pulsar://localhost:19092
+```
+
+In the above example,
+- `kafkaListener` is split into multiple tokens by a comma (`,`), the format of each token format is `<listener-name>://<host>:<port>`.
+- `kafkaProtocolMap` is split into multiple tokens by a comma (`,`), the format of each token format is `<listener-name>:<protocol>`.
+- `advertisedListeners` is split into multiple tokens by a comma(`,`), the format of each token format is `<listener-name>:<scheme>://<host>:<port>`.
+
+> **NOTE**
+>
+> In Pulsar, the `scheme` part could be `pulsar` or `pulsar+ssl`, but in KoP, the `scheme` part must be `pulsar`.
+
+
 ## Logger
 
 KoP shares the same configuration files with the Pulsar broker, e.g. `conf/broker.conf` or `conf/standalone.conf`. The log configurations can be configured in `conf/log4j2.yaml` file like below:

docs/kop.md

@@ -75,7 +75,7 @@ After you copy the `.nar` file to your Pulsar `/protocols` directory, you need t
     | `protocolHandlerDirectory`|./protocols  | Location of KoP NAR file |
     | `allowAutoTopicCreationType`| non-partitioned | partitioned |
 
-    By default, `allowAutoTopicCreationType` is set to `non-partitioned`. You need to set `allowAutoTopicCreationType` to `partitioned` because KoP only supports partitioned topics. If not, topics automatically created by KoP are still partitioned topics, yet topics created automatically by the Pulsar broker are non-partitioned topics.
+    By default, `allowAutoTopicCreationType` is set to `non-partitioned`. Since topics are partitioned by default in Kafka, it's better to avoid creating non-partitioned topics for Kafka clients unless Kafka clients need to interact with existing non-partitioned topics.
 
 2. Set Kafka listeners.
 

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

@@ -202,13 +202,6 @@ public class KafkaServiceConfiguration extends ServiceConfiguration {
     )
     private String kafkaAdvertisedListeners;
 
-    @Deprecated
-    @FieldContext(
-            category = CATEGORY_KOP,
-            doc = "Use kafkaProtocolMap, kafkaListeners and advertisedAddress instead."
-    )
-    private String kafkaListenerName;
-
     @FieldContext(
             category = CATEGORY_KOP,
             doc = "limit the queue size for request, \n"

kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfigurationTest.java

@@ -135,7 +135,6 @@ public void testConfigurationUtilsStream() throws Exception {
         assertEquals(kafkaServiceConfig.getManagedLedgerDigestType(), DigestType.CRC32C);
         assertEquals(
                 kafkaServiceConfig.getKopAllowedNamespaces(), Sets.newHashSet("public/default", "public/__kafka"));
-        assertEquals(kafkaServiceConfig.getKafkaListenerName(), "external");
     }
 
     @Test

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

@@ -52,7 +52,6 @@ public void testListenerName() throws Exception {
                 "pulsar:pulsar://" + localAddress + ":" + brokerPort
                         + ",kafka:pulsar://" + "localhost:" + kafkaBrokerPort;
         conf.setAdvertisedListeners(advertisedListeners);
-        conf.setKafkaListenerName("kafka");
         log.info("Set advertisedListeners to {}", advertisedListeners);
         super.internalSetup();