[storage] Fix Kafka TLS configuration when tls.enabled is missing

This change fixes the Kafka TLS configuration to work correctly when tls.enabled
flag is not provided but authentication=tls is set. Previously, TLS would not
be enabled in this case.

Changes:
- TLS is now properly configured when authentication=tls, regardless of tls.enabled
- Maintains backward compatibility with existing tls.enabled flag
- Sets explicit insecure mode only when TLS is intentionally disabled

Testing:
- Added unit tests for TLS configuration scenarios
- Verified with local Kafka cluster using TLS authentication
- Tested with HotROD example application

Resolves jaegertracing#6744

Signed-off-by: Amol Verma <[email protected]>

Author: Amol Verma

pkg/kafka/auth/config.go

@@ -43,11 +43,12 @@ func (config *AuthenticationConfig) SetConfiguration(saramaConfig *sarama.Config
 	if strings.Trim(authentication, " ") == "" {
 		authentication = none
 	}
-	if config.Authentication == tls {
-		if err := setTLSConfiguration(&config.TLS, saramaConfig, logger); err != nil {
-			return err
-		}
-	}
+   if config.Authentication == tls || !config.TLS.Insecure {
+        if err := setTLSConfiguration(&config.TLS, saramaConfig, logger); err != nil {
+            return err
+        }
+    }
+
 	switch authentication {
 	case none:
 		return nil
@@ -57,7 +58,7 @@ func (config *AuthenticationConfig) SetConfiguration(saramaConfig *sarama.Config
 		setKerberosConfiguration(&config.Kerberos, saramaConfig)
 		return nil
 	case plaintext:
-		return setPlainTextConfiguration(&config.PlainText, saramaConfig)
+        return setPlainTextConfiguration(&config.PlainText, saramaConfig)
 	default:
 		return fmt.Errorf("Unknown/Unsupported authentication method %s to kafka cluster", config.Authentication)
 	}
@@ -75,19 +76,22 @@ func (config *AuthenticationConfig) InitFromViper(configPrefix string, v *viper.
 	config.Kerberos.KeyTabPath = v.GetString(configPrefix + kerberosPrefix + suffixKerberosKeyTab)
 	config.Kerberos.DisablePAFXFast = v.GetBool(configPrefix + kerberosPrefix + suffixKerberosDisablePAFXFAST)
 
-	if config.Authentication == tls {
-		if !v.IsSet(configPrefix + ".tls.enabled") {
-			v.Set(configPrefix+".tls.enabled", "true")
-		}
-		tlsClientConfig := tlscfg.ClientFlagsConfig{
-			Prefix: configPrefix,
-		}
-		tlsCfg, err := tlsClientConfig.InitFromViper(v)
-		if err != nil {
-			return fmt.Errorf("failed to process Kafka TLS options: %w", err)
-		}
-		config.TLS = tlsCfg
-	}
+	if config.Authentication == tls || v.GetBool(configPrefix+".tls.enabled"){
+        tlsClientConfig := tlscfg.ClientFlagsConfig{
+            Prefix: configPrefix,
+        }
+        tlsCfg, err := tlsClientConfig.InitFromViper(v)
+        if err != nil {
+            return fmt.Errorf("failed to process Kafka TLS options: %w", err)
+        }
+        config.TLS = tlsCfg
+	} else {
+        // Explicitly set TLS to insecure when disabled
+        config.TLS = configtls.ClientConfig{
+            Insecure: true,
+        }
+    }
+
 
 	config.PlainText.Username = v.GetString(configPrefix + plainTextPrefix + suffixPlainTextUsername)
 	config.PlainText.Password = v.GetString(configPrefix + plainTextPrefix + suffixPlainTextPassword)

pkg/kafka/auth/tls.go

@@ -12,14 +12,17 @@ import (
 	"go.uber.org/zap"
 )
 
-func setTLSConfiguration(config *configtls.ClientConfig, saramaConfig *sarama.Config, _ *zap.Logger) error {
-	if !config.Insecure {
-		tlsConfig, err := config.LoadTLSConfig(context.Background())
-		if err != nil {
-			return fmt.Errorf("error loading tls config: %w", err)
-		}
-		saramaConfig.Net.TLS.Enable = true
-		saramaConfig.Net.TLS.Config = tlsConfig
-	}
-	return nil
-}
+func setTLSConfiguration(config *configtls.ClientConfig, saramaConfig *sarama.Config, logger *zap.Logger) error {
+    tlsConfig, err := config.LoadTLSConfig(context.Background())
+    if err != nil {
+        return fmt.Errorf("error loading tls config: %w", err)
+    }
+    
+    saramaConfig.Net.TLS.Enable = true
+    saramaConfig.Net.TLS.Config = tlsConfig
+    logger.Info("TLS configuration enabled for Kafka client", 
+        zap.Bool("skip_verify", config.InsecureSkipVerify),
+        zap.String("ca_file", config.CAFile),
+        zap.Bool("system_ca_enabled", config.Config.IncludeSystemCACertsPool))
+    return nil
+}