add config for allowAutoTopicCreationType=partitioned in doc (apache#132

) * fix log format * add config for allowAutoTopicCreationType=partitioned in doc
hangc0276 · Mar 29, 2020 · 9d79d6a · 9d79d6a
1 parent c6b5d43
commit 9d79d6a
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 6 deletions.
diff --git a/DEVELOP.md b/DEVELOP.md
@@ -59,18 +59,22 @@ As mentioned previously, KoP module is loaded along with the Pulsar broker. You
 1. Set the configuration of the KoP protocol handler.
 
     Add the following properties and set their values in Pulsar configuration file, such as `conf/broker.conf` or `conf/standalone.conf`.
-
+
+    Regarding topic auto create partition type, if you are not using [StreamNative Platform](https://streamnative.io/docs/v1.0.0/), please set it to `partitioned`.
 
     Property | Set it to the following value | Default value
     |---|---|---
     `messagingProtocols` | kafka | null
     `protocolHandlerDirectory`| Location of KoP NAR file | ./protocols
-
+    `protocolHandlerDirectory`| Location of KoP NAR file | ./protocols
+    `allowAutoTopicCreationType`| partitioned | non-partitioned
+
     **Example**
 
     ```
     messagingProtocols=kafka
     protocolHandlerDirectory=./protocols
+    allowAutoTopicCreationType=partitioned
     ```
 
 2. Set Kafka service listeners.

diff --git a/README.md b/README.md
@@ -64,18 +64,22 @@ add configurations in Pulsar's configuration file, such as `broker.conf` or `sta
 1. Set the configuration of the KoP protocol handler.
 
     Add the following properties and set their values in Pulsar configuration file, such as `conf/broker.conf` or `conf/standalone.conf`.
-
+
+    Regarding topic auto create partition type, if you are not using [StreamNative Platform](https://streamnative.io/docs/v1.0.0/), please set it to `partitioned`.
 
     Property | Set it to the following value | Default value
     |---|---|---
     `messagingProtocols` | kafka | null
     `protocolHandlerDirectory`| Location of KoP NAR file | ./protocols
-
+    `protocolHandlerDirectory`| Location of KoP NAR file | ./protocols
+    `allowAutoTopicCreationType`| partitioned | non-partitioned
+
     **Example**
 
     ```
     messagingProtocols=kafka
     protocolHandlerDirectory=./protocols
+    allowAutoTopicCreationType=partitioned
     ```
 
 2. Set Kafka service listeners.

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
@@ -1343,8 +1343,8 @@ private CompletableFuture<PartitionMetadata> findBroker(TopicName topic) {
                     && localListeners.contains(kopBrokerUrl)) {
                     topicManager.getTopic(topic.toString()).whenComplete((persistentTopic, exception) -> {
                         if (exception != null || persistentTopic == null) {
-                            log.warn("[{}] findBroker: Failed to getOrCreateTopic {}. exception:",
-                                ctx.channel(), topic.toString(), exception);
+                            log.warn("[{}] findBroker: Failed to getOrCreateTopic {}. broker:{}, exception:",
+                                ctx.channel(), topic.toString(), kopBrokerUrl, exception);
                             returnFuture.complete(null);
                         } else {
                             if (log.isDebugEnabled()) {