diff --git a/docs/configuration.md b/docs/configuration.md index 630091d477..c4d1865d95 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -5,14 +5,34 @@ | Name | Description | | ------------------------ | ------------------------------------------------------------ | | kafkaListeners | Comma-separated list of URIs that we will listen on and the listener names.
e.g. PLAINTEXT://localhost:9092,SSL://localhost:9093.
If the hostname is not set, the default interface is used. | +| kafkaProtocolMap | Comma-separated map of listener name and protocol.
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.
The format is the same as `kafkaListeners`. | -| kafkaListenerName | Specify the internal listener name for the broker.
The listener name must be contained in the advertisedListeners.
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 `://:`. +- `kafkaProtocolMap` is split into multiple tokens by a comma (`,`), the format of each token format is `:`. +- `advertisedListeners` is split into multiple tokens by a comma(`,`), the format of each token format is `:://:`. + +> **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: diff --git a/docs/kop.md b/docs/kop.md index 3bf87d7664..118277c5d0 100644 --- a/docs/kop.md +++ b/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. diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java index 174ba15bd2..949c8644de 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java +++ b/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" diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfigurationTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfigurationTest.java index 5f51b175a4..6296364689 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfigurationTest.java +++ b/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 diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java index 834f22a41e..ec1023b894 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java +++ b/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();