Expose GSSAPI Service name configuration of Kafka Scaler

CHANGELOG.md

@@ -89,6 +89,7 @@ New deprecation(s):
 - **General**: Introduce ENABLE_OPENTELEMETRY in deploying/testing process  ([#5375](https://github.com/kedacore/keda/issues/5375))
 - **General**: Migrate away from unmaintained golang/mock and use uber/gomock ([#5440](https://github.com/kedacore/keda/issues/5440))
 - **General**: Minor refactor to reduce copy/paste code in ScaledObject webhook ([#5397](https://github.com/kedacore/keda/issues/5397))
+- **Kafka**: Expose GSSAPI service name  ([#5474](https://github.com/kedacore/keda/issues/5474))
 
 ## v2.13.1
 

pkg/scalers/kafka_scaler.go

@@ -75,9 +75,10 @@ type kafkaMetadata struct {
 	password string
 
 	// GSSAPI
-	keytabPath         string
-	realm              string
-	kerberosConfigPath string
+	keytabPath          string
+	realm               string
+	kerberosConfigPath  string
+	kerberosServiceName string
 
 	// OAUTHBEARER
 	scopes                []string
@@ -291,6 +292,10 @@ func parseKerberosParams(config *scalersconfig.ScalerConfig, meta *kafkaMetadata
 	}
 	meta.kerberosConfigPath = path
 
+	if config.AuthParams["kerberosServiceName"] != "" {
+		meta.kerberosServiceName = strings.TrimSpace(config.AuthParams["kerberosServiceName"])
+	}
+
 	meta.saslType = mode
 	return nil
 }
@@ -541,7 +546,11 @@ func getKafkaClients(metadata kafkaMetadata) (sarama.Client, sarama.ClusterAdmin
 	if metadata.saslType == KafkaSASLTypeGSSAPI {
 		config.Net.SASL.Enable = true
 		config.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
-		config.Net.SASL.GSSAPI.ServiceName = "kafka"
+		if metadata.kerberosServiceName != "" {
+			config.Net.SASL.GSSAPI.ServiceName = metadata.kerberosServiceName
+		} else {
+			config.Net.SASL.GSSAPI.ServiceName = "kafka"
+		}
 		config.Net.SASL.GSSAPI.Username = metadata.username
 		config.Net.SASL.GSSAPI.Realm = metadata.realm
 		config.Net.SASL.GSSAPI.KerberosConfigPath = metadata.kerberosConfigPath

pkg/scalers/kafka_scaler_test.go

@@ -161,6 +161,8 @@ var parseKafkaAuthParamsTestDataset = []parseKafkaAuthParamsTestData{
 	{map[string]string{"sasl": "gssapi", "username": "admin", "password": "admin", "kerberosConfig": "<config>", "realm": "tst.com", "tls": "enable", "ca": "caaa", "cert": "ceert", "key": "keey"}, false, true},
 	// success, SASL GSSAPI/keytab + TLS
 	{map[string]string{"sasl": "gssapi", "username": "admin", "keytab": "/path/to/keytab", "kerberosConfig": "<config>", "realm": "tst.com", "tls": "enable", "ca": "caaa", "cert": "ceert", "key": "keey"}, false, true},
+	// success, SASL GSSAPI, KerberosServiceName supported
+	{map[string]string{"sasl": "gssapi", "username": "admin", "keytab": "/path/to/keytab", "kerberosConfig": "<config>", "realm": "tst.com", "kerberosServiceName": "srckafka"}, false, false},
 	// failure, SASL OAUTHBEARER + TLS bad sasl type
 	{map[string]string{"sasl": "foo", "username": "admin", "password": "admin", "scopes": "scope", "oauthTokenEndpointUri": "https://website.com", "tls": "disable"}, true, false},
 	// success, SASL OAUTHBEARER + TLS missing scope
@@ -412,6 +414,9 @@ func TestKafkaAuthParamsInTriggerAuthentication(t *testing.T) {
 					t.Errorf(err.Error())
 				}
 			}
+			if meta.kerberosServiceName != testData.authParams["kerberosServiceName"] {
+				t.Errorf("Expected kerberos ServiceName to be set to %v but got %v\n", testData.authParams["kerberosServiceName"], meta.kerberosServiceName)
+			}
 		}
 	}
 }