From 65a3601c6ec2dcded8ab31988f678a75f8102fa2 Mon Sep 17 00:00:00 2001
From: Egor <31911894+paragor@users.noreply.github.com>
Date: Fri, 30 Jun 2023 09:03:21 +0700
Subject: [PATCH] Add CaFile for KafkaRecorder when simpleSSL enabled (#534)

---
 pkg/handler/data_recorder_kafka.go | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/pkg/handler/data_recorder_kafka.go b/pkg/handler/data_recorder_kafka.go
index dc07acc7..f3dad018 100644
--- a/pkg/handler/data_recorder_kafka.go
+++ b/pkg/handler/data_recorder_kafka.go
@@ -92,23 +92,14 @@ var NewKafkaRecorder = func() DataRecorder {
 }
 
 func createTLSConfiguration(certFile string, keyFile string, caFile string, verifySSL bool, simpleSSL bool) (t *tls.Config) {
-	if certFile != "" && keyFile != "" && caFile != "" {
+	if certFile != "" && keyFile != "" {
 		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
 		if err != nil {
 			logrus.WithField("TLSConfigurationError", err).Panic(err)
 		}
 
-		caCert, err := os.ReadFile(caFile)
-		if err != nil {
-			logrus.WithField("TLSConfigurationError", err).Panic(err)
-		}
-
-		caCertPool := x509.NewCertPool()
-		caCertPool.AppendCertsFromPEM(caCert)
-
 		t = &tls.Config{
 			Certificates:       []tls.Certificate{cert},
-			RootCAs:            caCertPool,
 			InsecureSkipVerify: !verifySSL,
 		}
 	}
@@ -118,6 +109,17 @@ func createTLSConfiguration(certFile string, keyFile string, caFile string, veri
 			InsecureSkipVerify: !verifySSL,
 		}
 	}
+
+	if caFile != "" && t != nil {
+		caCert, err := os.ReadFile(caFile)
+		if err != nil {
+			logrus.WithField("TLSConfigurationError", err).Panic(err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+		t.RootCAs = caCertPool
+	}
 	// will be nil by default if nothing is provided
 	return t
 }