kafka: allow disabling FAST in sarama client

Our sarama client has kerberos FAST negotiation turned on by default,
but there are KDCs that can't handle FAST negotiation and will fail.
There is an option to configure this on the sarama client, but we didn't
expose it anywhere, so users couldn't get to it.

This just adds an additional auth parameter to AuthConfig to expose that
configuration option so users who need to shut off FAST are able to do
so.

Signed-off-by: John Kyros <[email protected]>

Author: jkyros

CHANGELOG.md

@@ -73,6 +73,7 @@ Here is an overview of all new **experimental** features:
 - **GCP Scalers**: Added custom time horizon in GCP scalers ([#5778](https://github.com/kedacore/keda/issues/5778))
 - **GitHub Scaler**: Fixed pagination, fetching repository list ([#5738](https://github.com/kedacore/keda/issues/5738))
 - **Kafka**: Fix logic to scale to zero on invalid offset even with earliest offsetResetPolicy ([#5689](https://github.com/kedacore/keda/issues/5689))
+- **Kafka**: Allow disabling FAST negotation when using Kerberos ([#6188](https://github.com/kedacore/keda/issues/6188))
 - **RabbitMQ Scaler**: Add connection name for AMQP ([#5958](https://github.com/kedacore/keda/issues/5958))
 - TODO ([#XXX](https://github.com/kedacore/keda/issues/XXX))
 

pkg/scalers/kafka_scaler.go

@@ -80,6 +80,7 @@ type kafkaMetadata struct {
 	realm               string
 	kerberosConfigPath  string
 	kerberosServiceName string
+	kerberosDisableFAST bool
 
 	// OAUTHBEARER
 	tokenProvider         kafkaSaslOAuthTokenProvider
@@ -408,6 +409,15 @@ func parseKerberosParams(config *scalersconfig.ScalerConfig, meta *kafkaMetadata
 		meta.kerberosServiceName = strings.TrimSpace(config.AuthParams["kerberosServiceName"])
 	}
 
+	meta.kerberosDisableFAST = false
+	if val, ok := config.AuthParams["kerberosDisableFAST"]; ok {
+		t, err := strconv.ParseBool(val)
+		if err != nil {
+			return fmt.Errorf("error parsing kerberosDisableFAST: %w", err)
+		}
+		meta.kerberosDisableFAST = t
+	}
+
 	meta.saslType = mode
 	return nil
 }
@@ -687,7 +697,12 @@ func getKafkaClientConfig(ctx context.Context, metadata kafkaMetadata) (*sarama.
 			config.Net.SASL.GSSAPI.AuthType = sarama.KRB5_USER_AUTH
 			config.Net.SASL.GSSAPI.Password = metadata.password
 		}
+
+		if metadata.kerberosDisableFAST {
+			config.Net.SASL.GSSAPI.DisablePAFXFAST = true
+		}
 	}
+
 	return config, nil
 }
 

pkg/scalers/kafka_scaler_test.go

@@ -209,6 +209,10 @@ var parseKafkaAuthParamsTestDataset = []parseKafkaAuthParamsTestData{
 	{map[string]string{"sasl": "gssapi", "username": "admin", "password": "admin", "kerberosConfig": "<config>", "tls": "enable", "ca": "caaa", "cert": "ceert", "key": "keey"}, true, false},
 	// failure, SASL GSSAPI/keytab + TLS missing username
 	{map[string]string{"sasl": "gssapi", "keytab": "/path/to/keytab", "kerberosConfig": "<config>", "realm": "tst.com", "tls": "enable", "ca": "caaa", "cert": "ceert", "key": "keey"}, true, false},
+	// success, SASL GSSAPI/disableFast
+	{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},
 }
 var parseAuthParamsTestDataset = []parseAuthParamsTestDataSecondAuthMethod{
 	// success, SASL plaintext