MINOR: fix failing system test delegation_token_test

[rondagostino]

The system test in delegation_token_test.py broke due to #10199.
That patch changed the logic of SecurityConfig.enabled_sasl_mechanisms() to only add the inter-broker SASL mechanism when the inter-broker protocol was SASL_{PLAINTEXT,SSL}. The inter-broker protocol is PLAINTEXT in delegation_token_test.py, so the default inter-broker SASL mechanism of GSSAPI was not being added into the set returned by enabled_sasl_mechanisms(). This is actually correct -- it shouldn't be added if it isn't used for inter-broker communication. It should be added because clients use it, of course -- SASL_PLAINTEXT is the security protocol on an advertised listener, and client_sasl_mechanism is set to the .csv value "GSSAPI,SCRAM-SHA-256"in delegation_token_test. Unfortunately in #10199 we did not take into account the possibility that client_sasl_mechanism could be a .csv value, and we therefore fail to create a krb5.conf file, which causes kafka-delegation_tokens.sh to fail. This bug of .csv omission therefore uncovered a different bug -- we were relying on the default inter-broker SASL mechanism to signal that Kerberos was being used even though the inter-broker protocol wasn't SASL. This patch explicitly includes the elements of the client_sasl_mechanism .csv value (which in most cases is just a single value but in delegation_token_test it is not).

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[mumrah]

Not sure if we should bother in this PR, but the usages of client_sasl_mechanism could stand to be cleaned up. In SecurityConfig.__init__ we default it to a simple string, but as you found here (and as seen in SecurityConfig.client_config) we support it being a comma delimited string.

kafka/tests/kafkatest/services/security/security_config.py

Lines 240 to 253 in 58b3b1b

	if use_inter_broker_mechanism_for_client:
	client_sasl_mechanism_to_use = self.interbroker_sasl_mechanism
	else:
	# csv is supported here, but client configs only supports a single mechanism,
	# so arbitrarily take the first one defined in case it has multiple values
	client_sasl_mechanism_to_use = self.client_sasl_mechanism.split(',')[0].strip()
	return SecurityConfig(self.context, self.security_protocol,
	client_sasl_mechanism=client_sasl_mechanism_to_use,
	template_props=template_props,
	static_jaas_conf=static_jaas_conf,
	jaas_override_variables=jaas_override_variables,
	listener_security_config=self.listener_security_config,
	tls_version=self.tls_version)

It's probably a lot safer to declare this as a list in the class and not worry about having to do the split(",") everywhere. Though maybe there's a reason why we split it lazily.. not sure.

Either way, this change looks good. However, you might consider doing something like:

sasl_mechanisms += [mechanism.strip() for mechanism in self.client_sasl_mechanism.split(',')]

since self.client_sasl_mechanism.strip() won't catch spaces in the middle of the string.

[rondagostino]

Not sure if we should bother in this PR, but the usages of client_sasl_mechanism could stand to be cleaned up

I agree it needs to be cleaned up. Given we are past code freeze for 2.8, I've opened https://issues.apache.org/jira/browse/KAFKA-12402 for this and we can address it another time.

[cmccabe]

LGTM