From 8265969f523cddb601211e4f070fef4a56b7aae0 Mon Sep 17 00:00:00 2001
From: Gerald Quintana <gerald.quintana@gmail.com>
Date: Thu, 21 Oct 2021 23:32:39 +0200
Subject: [PATCH] feat: Kafka Add metadata full to config (#9833)

---
 etc/telegraf.conf                               | 6 ++++++
 plugins/common/kafka/config.go                  | 8 ++++++++
 plugins/inputs/kafka_consumer/README.md         | 3 +++
 plugins/inputs/kafka_consumer/kafka_consumer.go | 3 +++
 plugins/outputs/kafka/README.md                 | 3 +++
 plugins/outputs/kafka/kafka.go                  | 3 +++
 6 files changed, 26 insertions(+)

diff --git a/etc/telegraf.conf b/etc/telegraf.conf
index 2b09df662..ae5680b32 100644
--- a/etc/telegraf.conf
+++ b/etc/telegraf.conf
@@ -1081,6 +1081,9 @@
 #   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
 #   # sasl_version = 1
 #
+#   # Disable Kafka metadata full fetch
+#   # metadata_full = false
+#
 #   ## Data format to output.
 #   ## Each data format has its own unique set of configuration options, read
 #   ## more about them here:
@@ -7618,6 +7621,9 @@
 #   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
 #   # sasl_version = 1
 #
+#   # Disable Kafka metadata full fetch
+#   # metadata_full = false
+#
 #   ## Name of the consumer group.
 #   # consumer_group = "telegraf_metrics_consumers"
 #
diff --git a/plugins/common/kafka/config.go b/plugins/common/kafka/config.go
index 56e70a26b..432d57552 100644
--- a/plugins/common/kafka/config.go
+++ b/plugins/common/kafka/config.go
@@ -55,6 +55,9 @@ type Config struct {
 
 	// EnableTLS deprecated
 	EnableTLS *bool `toml:"enable_tls"`
+
+	// Disable full metadata fetching
+	MetadataFull *bool `toml:"metadata_full"`
 }
 
 // SetConfig on the sarama.Config object from the Config struct.
@@ -89,5 +92,10 @@ func (k *Config) SetConfig(config *sarama.Config) error {
 		config.Net.TLS.Enable = true
 	}
 
+	if k.MetadataFull != nil {
+		// Defaults to true in Sarama
+		config.Metadata.Full = *k.MetadataFull
+	}
+
 	return k.SetSASLConfig(config)
 }
diff --git a/plugins/inputs/kafka_consumer/README.md b/plugins/inputs/kafka_consumer/README.md
index ac04925a2..741f24d04 100644
--- a/plugins/inputs/kafka_consumer/README.md
+++ b/plugins/inputs/kafka_consumer/README.md
@@ -59,6 +59,9 @@ and use the old zookeeper connection method.
   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
   # sasl_version = 1
 
+  # Disable Kafka metadata full fetch
+  # metadata_full = false
+
   ## Name of the consumer group.
   # consumer_group = "telegraf_metrics_consumers"
 
diff --git a/plugins/inputs/kafka_consumer/kafka_consumer.go b/plugins/inputs/kafka_consumer/kafka_consumer.go
index c6894fd74..70affdc23 100644
--- a/plugins/inputs/kafka_consumer/kafka_consumer.go
+++ b/plugins/inputs/kafka_consumer/kafka_consumer.go
@@ -66,6 +66,9 @@ const sampleConfig = `
   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
   # sasl_version = 1
 
+  # Disable Kafka metadata full fetch
+  # metadata_full = false
+
   ## Name of the consumer group.
   # consumer_group = "telegraf_metrics_consumers"
 
diff --git a/plugins/outputs/kafka/README.md b/plugins/outputs/kafka/README.md
index e76522018..54108d8be 100644
--- a/plugins/outputs/kafka/README.md
+++ b/plugins/outputs/kafka/README.md
@@ -136,6 +136,9 @@ This plugin writes to a [Kafka Broker](http://kafka.apache.org/07/quickstart.htm
   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
   # sasl_version = 1
 
+  # Disable Kafka metadata full fetch
+  # metadata_full = false
+
   ## Data format to output.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:
diff --git a/plugins/outputs/kafka/kafka.go b/plugins/outputs/kafka/kafka.go
index d30c730cf..297242700 100644
--- a/plugins/outputs/kafka/kafka.go
+++ b/plugins/outputs/kafka/kafka.go
@@ -212,6 +212,9 @@ var sampleConfig = `
   ## SASL protocol version.  When connecting to Azure EventHub set to 0.
   # sasl_version = 1
 
+  # Disable Kafka metadata full fetch
+  # metadata_full = false
+
   ## Data format to output.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here: