feat(kafka): Add keep-alive period setting for input and output. (#12459)

plugins/common/kafka/config.go

@@ -54,10 +54,11 @@ type Config struct {
 	SASLAuth
 	tls.ClientConfig
 
-	Version          string `toml:"version"`
+	Version          string           `toml:"version"`
-	ClientID         string `toml:"client_id"`
+	ClientID         string           `toml:"client_id"`
-	CompressionCodec int    `toml:"compression_codec"`
+	CompressionCodec int              `toml:"compression_codec"`
-	EnableTLS        *bool  `toml:"enable_tls"`
+	EnableTLS        *bool            `toml:"enable_tls"`
+	KeepAlivePeriod  *tgConf.Duration `toml:"keep_alive_period"`
 
 	MetadataRetryMax         int             `toml:"metadata_retry_max"`
 	MetadataRetryType        string          `toml:"metadata_retry_type"`
@@ -118,6 +119,11 @@ func (k *Config) SetConfig(config *sarama.Config, log telegraf.Logger) error {
 		}
 	}
 
+	if k.KeepAlivePeriod != nil {
+		// Defaults to OS setting (15s currently)
+		config.Net.KeepAlive = time.Duration(*k.KeepAlivePeriod)
+	}
+
 	if k.MetadataFull != nil {
 		// Defaults to true in Sarama
 		config.Metadata.Full = *k.MetadataFull

plugins/inputs/kafka_consumer/README.md

@@ -45,6 +45,10 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Period between keep alive probes.
+  ## Defaults to the OS configuration if not specified or zero.
+  # keep_alive_period = "15s"
+
   ## SASL authentication credentials.  These settings should typically be used
   ## with TLS encryption enabled
   # sasl_username = "kafka"

plugins/inputs/kafka_consumer/sample.conf

@@ -25,6 +25,10 @@
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Period between keep alive probes.
+  ## Defaults to the OS configuration if not specified or zero.
+  # keep_alive_period = "15s"
+
   ## SASL authentication credentials.  These settings should typically be used
   ## with TLS encryption enabled
   # sasl_username = "kafka"

plugins/outputs/kafka/README.md

@@ -95,7 +95,7 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ##  2 : Snappy
   ##  3 : LZ4
   ##  4 : ZSTD
-   # compression_codec = 0
+  # compression_codec = 0
 
   ## Idempotent Writes
   ## If enabled, exactly one copy of each message is written.
@@ -133,6 +133,10 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Period between keep alive probes.
+  ## Defaults to the OS configuration if not specified or zero.
+  # keep_alive_period = "15s"
+
   ## Optional SOCKS5 proxy to use when connecting to brokers
   # socks5_enabled = true
   # socks5_address = "127.0.0.1:1080"

plugins/outputs/kafka/sample.conf

@@ -78,7 +78,7 @@
   ##  2 : Snappy
   ##  3 : LZ4
   ##  4 : ZSTD
-   # compression_codec = 0
+  # compression_codec = 0
 
   ## Idempotent Writes
   ## If enabled, exactly one copy of each message is written.
@@ -116,6 +116,10 @@
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Period between keep alive probes.
+  ## Defaults to the OS configuration if not specified or zero.
+  # keep_alive_period = "15s"
+
   ## Optional SOCKS5 proxy to use when connecting to brokers
   # socks5_enabled = true
   # socks5_address = "127.0.0.1:1080"