From aedf58d74c8d68b6d0855336e2bf8c7dd0706d90 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sun, 26 Sep 2021 18:32:06 +0800 Subject: [PATCH 1/5] Update docs for allowAutoTopicCreationType --- docs/kop.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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. From 674cbe3627d1144e31164ee4b57fbd6946c6d116 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 28 Sep 2021 17:17:02 +0800 Subject: [PATCH 2/5] Remove kafkaListenerName --- docs/configuration.md | 1 - .../pulsar/handlers/kop/KafkaServiceConfiguration.java | 7 ------- .../pulsar/handlers/kop/KafkaServiceConfigurationTest.java | 1 - .../pulsar/handlers/kop/KafkaListenerNameTest.java | 1 - 4 files changed, 10 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 630091d477..bd7607ec1f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -7,7 +7,6 @@ | 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. | | 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. | > **NOTE** > 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(); From daddac759a3fd1e73974ecbd71bb2c8c32f9b64b Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 28 Sep 2021 17:37:26 +0800 Subject: [PATCH 3/5] Add docs for multiple listeners --- docs/configuration.md | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index bd7607ec1f..56a02fb675 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -5,13 +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`. | +| 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 name 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're going 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 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 above example, +- `kafkaListener` is split into multiple tokens by `,`, each token's format is `://:`. +- `kafkaProtocolMap` is split into multiple tokens by `,`, each token's format is `:`. +- `advertisedListeners` is split into multiple tokens by `,`, each token's 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: From 5a88f188dc0ba9e876be8b8a01b8357219deddb1 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 28 Sep 2021 20:18:51 +0800 Subject: [PATCH 4/5] Address comments --- docs/configuration.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 56a02fb675..0110a58082 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -13,9 +13,9 @@ > > Among all configurations, only `kafkaListeners` or `listeners` (deprecated) is required. -To support multiple listeners, you need to specify different listener name 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`. +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're going 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 following configurations: +For example, assuming you're going 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 From 0133d0835409e8c08e324458c57d4c246010ac14 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 30 Sep 2021 14:36:45 +0800 Subject: [PATCH 5/5] Address comments --- docs/configuration.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 0110a58082..c4d1865d95 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -15,7 +15,7 @@ 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're going 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: +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 @@ -23,10 +23,10 @@ kafkaProtocolMap=kafka_internal:PLAINTEXT,kafka_external:PLAINTEXT advertisedListeners=pulsar:pulsar://localhost:6650,kafka_internal:pulsar://localhost:9092,kafka_external:pulsar://localhost:19092 ``` -In above example, -- `kafkaListener` is split into multiple tokens by `,`, each token's format is `://:`. -- `kafkaProtocolMap` is split into multiple tokens by `,`, each token's format is `:`. -- `advertisedListeners` is split into multiple tokens by `,`, each token's format is `:://:`. +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** >