feat(kafka): honor allow.auto.create.topics (#1741)

Author: yang-xiaodong

docs/content/user-guide/en/transport/kafka.md

@@ -64,6 +64,18 @@ services.AddCap(capOptions =>
 
 [https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)
 
+To prevent CAP from creating topics automatically, disable topic auto creation:
+
+```csharp
+services.AddCap(capOptions =>
+{
+    capOptions.UseKafka(kafkaOption =>
+    {
+        kafkaOption.MainConfig.Add("allow.auto.create.topics", "false");
+    });
+});
+```
+
 #### CustomHeadersBuilder Options
 
 When the message sent from a heterogeneous system, because of the CAP needs to define additional headers, so an exception will occur at this time. By providing this parameter to set the custom headersn to make the subscriber works.

docs/content/user-guide/zh/transport/kafka.md

@@ -65,6 +65,18 @@ MainConfig 为配置字典，你可以通过以下链接找到其支持的配置
 
 [https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)
 
+要禁止 CAP 自动创建主题，可以关闭该功能：
+
+```csharp
+services.AddCap(capOptions =>
+{
+    capOptions.UseKafka(kafkaOption =>
+    {
+        kafkaOption.MainConfig.Add("allow.auto.create.topics", "false");
+    });
+});
+```
+
 #### CustomHeadersBuilder Options
 
 有关 `CustomHeadersBuilder` 的说明：

src/DotNetCore.CAP.Kafka/KafkaConsumerClient.cs

@@ -49,30 +49,41 @@ public ICollection<string> FetchTopics(IEnumerable<string> topicNames)
 
         var regexTopicNames = topicNames.Select(Helper.WildcardToRegex).ToList();
 
-        try
+        var allowAutoCreate = true;
+        if (_kafkaOptions.MainConfig.TryGetValue("allow.auto.create.topics", out var autoCreateValue)
+            && bool.TryParse(autoCreateValue, out var parsedValue))
         {
-            var config = new AdminClientConfig(_kafkaOptions.MainConfig) { BootstrapServers = _kafkaOptions.Servers };
+            allowAutoCreate = parsedValue;
+        }
+
+        if (allowAutoCreate)
+        {
+            try
+            {
+                var config = new AdminClientConfig(_kafkaOptions.MainConfig)
+                    { BootstrapServers = _kafkaOptions.Servers };
 
-            using var adminClient = new AdminClientBuilder(config).Build();
+                using var adminClient = new AdminClientBuilder(config).Build();
 
-            adminClient.CreateTopicsAsync(regexTopicNames.Select(x => new TopicSpecification
+                adminClient.CreateTopicsAsync(regexTopicNames.Select(x => new TopicSpecification
+                {
+                    Name = x,
+                    NumPartitions = _kafkaOptions.TopicOptions.NumPartitions,
+                    ReplicationFactor = _kafkaOptions.TopicOptions.ReplicationFactor
+                })).GetAwaiter().GetResult();
+            }
+            catch (CreateTopicsException ex) when (ex.Message.Contains("already exists"))
             {
-                Name = x,
-                NumPartitions = _kafkaOptions.TopicOptions.NumPartitions,
-                ReplicationFactor = _kafkaOptions.TopicOptions.ReplicationFactor
-            })).GetAwaiter().GetResult();
-        }
-        catch (CreateTopicsException ex) when (ex.Message.Contains("already exists"))
-        {
-        }
-        catch (Exception ex)
-        {
-            var logArgs = new LogMessageEventArgs
+            }
+            catch (Exception ex)
             {
-                LogType = MqLogType.ConsumeError,
-                Reason = "An error was encountered when automatically creating topic! -->" + ex.Message
-            };
-            OnLogCallback!(logArgs);
+                var logArgs = new LogMessageEventArgs
+                {
+                    LogType = MqLogType.ConsumeError,
+                    Reason = "An error was encountered when automatically creating topic! -->" + ex.Message
+                };
+                OnLogCallback!(logArgs);
+            }
         }
 
         return regexTopicNames;