Support TLS auth for Kafka notification target (#8609)

cmd/config/notify/help.go

@@ -163,6 +163,18 @@ var (
 			Optional:    true,
 			Type:        "number",
 		},
+		config.HelpKV{
+			Key:         target.KafkaClientTLSCert,
+			Description: "Set path to client certificate",
+			Optional:    true,
+			Type:        "path",
+		},
+		config.HelpKV{
+			Key:         target.KafkaClientTLSKey,
+			Description: "Set path to client key",
+			Optional:    true,
+			Type:        "path",
+		},
 		config.HelpKV{
 			Key:         config.Comment,
 			Description: config.DefaultComment,

cmd/config/notify/legacy.go

@@ -42,6 +42,14 @@ func SetNotifyKafka(s config.Config, kName string, cfg target.KafkaArgs) error {
 			Key:   target.KafkaQueueDir,
 			Value: cfg.QueueDir,
 		},
+		config.KV{
+			Key:   target.KafkaClientTLSCert,
+			Value: cfg.TLS.ClientTLSCert,
+		},
+		config.KV{
+			Key:   target.KafkaClientTLSKey,
+			Value: cfg.TLS.ClientTLSKey,
+		},
 		config.KV{
 			Key:   target.KafkaQueueLimit,
 			Value: strconv.Itoa(int(cfg.QueueLimit)),

cmd/config/notify/parse.go

@@ -360,6 +360,14 @@ var (
 			Key:   target.KafkaSASLPassword,
 			Value: "",
 		},
+		config.KV{
+			Key:   target.KafkaClientTLSCert,
+			Value: "",
+		},
+		config.KV{
+			Key:   target.KafkaClientTLSKey,
+			Value: "",
+		},
 		config.KV{
 			Key:   target.KafkaTLSClientAuth,
 			Value: "0",
@@ -467,10 +475,24 @@ func GetNotifyKafka(kafkaKVS map[string]config.KVS) (map[string]target.KafkaArgs
 		if k != config.Default {
 			tlsSkipVerifyEnv = tlsSkipVerifyEnv + config.Default + k
 		}
+
+		tlsClientTLSCertEnv := target.EnvKafkaClientTLSCert
+		if k != config.Default {
+			tlsClientTLSCertEnv = tlsClientTLSCertEnv + config.Default + k
+		}
+
+		tlsClientTLSKeyEnv := target.EnvKafkaClientTLSKey
+		if k != config.Default {
+			tlsClientTLSKeyEnv = tlsClientTLSKeyEnv + config.Default + k
+		}
+
 		kafkaArgs.TLS.Enable = env.Get(tlsEnableEnv, kv.Get(target.KafkaTLS)) == config.EnableOn
 		kafkaArgs.TLS.SkipVerify = env.Get(tlsSkipVerifyEnv, kv.Get(target.KafkaTLSSkipVerify)) == config.EnableOn
 		kafkaArgs.TLS.ClientAuth = tls.ClientAuthType(clientAuth)
 
+		kafkaArgs.TLS.ClientTLSCert = env.Get(tlsClientTLSCertEnv, kv.Get(target.KafkaClientTLSCert))
+		kafkaArgs.TLS.ClientTLSKey = env.Get(tlsClientTLSKeyEnv, kv.Get(target.KafkaClientTLSKey))
+
 		saslEnableEnv := target.EnvKafkaSASLEnable
 		if k != config.Default {
 			saslEnableEnv = saslEnableEnv + config.Default + k