fix: allowing usage of sasl none on kafka scaler (#7070)

* fix: allowing usage of sasl none on kafka scaler

Signed-off-by: Ronaldo Lanhellas <ronaldo.lanhellas@gmail.com>

* docs: allowing usage of sasl none on kafka scaler

Signed-off-by: Ronaldo Lanhellas <ronaldo.lanhellas@gmail.com>

* ci: allowing usage of sasl none on kafka scaler

Signed-off-by: Ronaldo Lanhellas <ronaldo.lanhellas@gmail.com>

* ci: allowing usage of sasl none on kafka scaler

Signed-off-by: Ronaldo Lanhellas <ronaldo.lanhellas@gmail.com>

---------

Signed-off-by: Ronaldo Lanhellas <ronaldo.lanhellas@gmail.com>

Author: rlanhellas

CHANGELOG.md

@@ -105,6 +105,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **General**: Sets hpaName in Status when ScaledObject adopts/finds an existing HPA ([#6336](https://github.com/kedacore/keda/issues/6336))
 - **Cron Scaler**: Fix cron scaler to return zero metric value by default([#6886](https://github.com/kedacore/keda/issues/6886))
 - **Hashicorp Vault**: Fix Vault PKI param using camel case param when making API call to Vault PKI endpoint ([#6864](https://github.com/kedacore/keda/issues/6864))
+- **Kafka Scaler**: Fix throwing error when using sasl=none ([#7061](https://github.com/kedacore/keda/issues/7061))
 - **Pulsar Scaler**: Resolve nil pointer dereference in Pulsar scaler redirect handling ([#7024](https://github.com/kedacore/keda/issues/7024))
 - **RabbitMQ Scaler**: Fix incorrect URL encoding in RabbitMQ vhosts containing %2f ([#6963](https://github.com/kedacore/keda/issues/6963))
 

pkg/scalers/kafka_scaler.go

@@ -222,10 +222,9 @@ func parseKafkaAuthParams(config *scalersconfig.ScalerConfig, meta *kafkaMetadat
 		saslAuthType = val
 	}
 
-	if saslAuthType != "" {
-		saslAuthType = strings.TrimSpace(saslAuthType)
-		mode := kafkaSaslType(saslAuthType)
-
+	saslAuthType = strings.TrimSpace(saslAuthType)
+	mode := kafkaSaslType(saslAuthType)
+	if saslAuthType != "" && mode != KafkaSASLTypeNone {
 		switch {
 		case mode == KafkaSASLTypePlaintext || mode == KafkaSASLTypeSCRAMSHA256 || mode == KafkaSASLTypeSCRAMSHA512:
 			err := parseSaslParams(config, meta, mode)

pkg/scalers/kafka_scaler_test.go

@@ -213,6 +213,8 @@ var parseKafkaAuthParamsTestDataset = []parseKafkaAuthParamsTestData{
 	{map[string]string{"sasl": "gssapi", "username": "admin", "keytab": "/path/to/keytab", "kerberosConfig": "<config>", "realm": "tst.com", "kerberosDisableFAST": "true"}, false, false},
 	// failure, SASL GSSAPI/disableFast incorrect
 	{map[string]string{"sasl": "gssapi", "username": "admin", "keytab": "/path/to/keytab", "kerberosConfig": "<config>", "realm": "tst.com", "kerberosDisableFAST": "notabool"}, true, false},
+	// success, SASL none
+	{map[string]string{"sasl": "none"}, false, false},
 }
 var parseAuthParamsTestDataset = []parseAuthParamsTestDataSecondAuthMethod{
 	// success, SASL plaintext
@@ -305,6 +307,8 @@ var parseAuthParamsTestDataset = []parseAuthParamsTestDataSecondAuthMethod{
 	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "version": "1.0.0"}, map[string]string{"sasl": "scram_sha512\n", "username": "admin", "password": "admin"}, false, true},
 	// success, setting SASL scram_sha512 value with extra space in TriggerAuthentication
 	{map[string]string{"bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "version": "1.0.0"}, map[string]string{"sasl": "scram_sha512 ", "username": "admin", "password": "admin"}, false, true},
+	// success, SASL none
+	{map[string]string{"sasl": "none", "bootstrapServers": "foobar:9092", "consumerGroup": "my-group", "topic": "my-topic", "allowIdleConsumers": "true", "version": "1.0.0"}, map[string]string{}, false, false},
 }
 
 var parseKafkaOAuthbearerAuthParamsTestDataset = []parseKafkaOAuthbearerAuthParamsTestData{