Ability to enable/disable Auto Topic Creation in Kafka input plugin. (#172)

The Logstash Kafka input plugin should have the ability to enable/disable the auto creation of topics from the Consumer (logstash).

Users who have enabled auto topic creation in Kafka so that topics are automatically created when a event is produced should also be allowed to enable OR disable the same from Logstash's end.

The downside of this autocreation function is that topics can also be created by a consumers. This makes it difficult to remove topics that no longer produce events. There wasn't a property for this in the logstash's Kafka input plugin documentation.

Documentation ref: https://cwiki.apache.org/confluence/display/KAFKA/KIP-361%3A+Add+Consumer+Configuration+to+Disable+Auto+Topic+Creation

Author: e-parth-pathak

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 11.5.0
+  - Add "auto_create_topics" option to allow disabling of topic auto creation [#172](https://github.com/logstash-plugins/logstash-integration-kafka/pull/172)
+
 ## 11.4.2
   - Add default client_id of logstash to kafka output [#169](https://github.com/logstash-plugins/logstash-integration-kafka/pull/169)
 

docs/input-kafka.asciidoc

@@ -98,6 +98,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 |=======================================================================
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-auto_commit_interval_ms>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-auto_create_topics>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-auto_offset_reset>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-bootstrap_servers>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-check_crcs>> |<<boolean,boolean>>|No
@@ -285,6 +286,17 @@ This will add a field named `kafka` to the logstash event containing the followi
 * `offset`: The offset from the partition this message is associated with
 * `key`: A ByteBuffer containing the message key
 
+
+[id="plugins-{type}s-{plugin}-auto_create_topics"]
+===== `auto_create_topics` 
+
+  * Value type is <<boolean,boolean>>
+  * Default value is `true`
+
+Controls whether the topic is automatically created when subscribing to a non-existent topic.
+A topic will be auto-created only if this configuration is set to `true` and auto-topic creation is enabled on the broker using `auto.create.topics.enable`;
+otherwise auto-topic creation is not permitted. 
+
 [id="plugins-{type}s-{plugin}-enable_auto_commit"]
 ===== `enable_auto_commit` 
 
@@ -789,7 +801,6 @@ Filtering by a regular expression is done by retrieving the full list of topic n
 NOTE:  When the broker has some topics configured with ACL rules and they miss the DESCRIBE permission, then the subscription
 happens but on the broker side it is logged that the subscription of some topics was denied to the configured user.
 
-
 [id="plugins-{type}s-{plugin}-value_deserializer_class"]
 ===== `value_deserializer_class` 
 

lib/logstash/inputs/kafka.rb

@@ -246,6 +246,12 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   #   `timestamp`: The timestamp of this message
   # While with `extended` it adds also all the key values present in the Kafka header if the key is valid UTF-8 else
   # silently skip it.
+  #
+  # Controls whether a kafka topic is automatically created when subscribing to a non-existent topic.
+  # A topic will be auto-created only if this configuration is set to `true` and auto-topic creation is enabled on the broker using `auto.create.topics.enable`; 
+  # otherwise auto-topic creation is not permitted.
+  config :auto_create_topics, :validate => :boolean, :default => true
+
   config :decorate_events, :validate => %w(none basic extended false true), :default => "none"
 
   attr_reader :metadata_mode
@@ -410,6 +416,7 @@ def create_consumer(client_id, group_instance_id)
 
       props.put(kafka::AUTO_COMMIT_INTERVAL_MS_CONFIG, auto_commit_interval_ms.to_s) unless auto_commit_interval_ms.nil?
       props.put(kafka::AUTO_OFFSET_RESET_CONFIG, auto_offset_reset) unless auto_offset_reset.nil?
+      props.put(kafka::ALLOW_AUTO_CREATE_TOPICS_CONFIG, auto_create_topics) unless auto_create_topics.nil?
       props.put(kafka::BOOTSTRAP_SERVERS_CONFIG, bootstrap_servers)
       props.put(kafka::CHECK_CRCS_CONFIG, check_crcs.to_s) unless check_crcs.nil?
       props.put(kafka::CLIENT_DNS_LOOKUP_CONFIG, client_dns_lookup)

logstash-integration-kafka.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-integration-kafka'
-  s.version         = '11.4.2'
+  s.version         = '11.5.0'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Integration with Kafka - input and output plugins"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline "+