...

The implementation is designed in such a way that it does not preclude adding support for re-authentication of other SASL mechanism (e.g. PLAIN, SCRAM-related, and GSSAPI), but doing so is explicitly out-of-scope for this proposal (a section below discusses how this could be done).  Also explicitly out-of-scope for this proposal is the ability for brokers to close connections that continue to use expired credentials.  This ability is a natural next step, but it will be addressed via a separate KIP if/when this one is adopted.

...

Finally, we add methods on SaslServerAuthenticator to respond to the requests that arrive related to re-authentication – specifically, respondToReauthenticationSaslHandshakeRequest() and respondToReauthenticationSaslAuthenticateRequest() – and we add code to kafka.server.KafkaApis to route the received requests accordingly (as opposed to responding with an error, which is what currently happens).  I won't go into the details here; see the PR for the code.

How To Support Re-Authentication for Other SASL Mechanisms

It was mentioned above that the design supports adding re-authentication to PLAIN, GSSAPI, and SCRAM-related SASL mechanisms but that doing so is out of scope for this KIP.  It is helpful to describe how this could be done in a bit more detail.

There are two requirements: the credential that the Login mechanism adds to the Subject's private credentials on the client side must implement the ExpiringCredential interface; and the client and server sides have to agree on the expiration time of each credential so that the client will know when to refresh the credential and the server will know when the presented credential expires (assuming we implement the ability for a server to close "expired" connections).

Implementing the ExpiringCredential interface is pretty easily done by wrapping the data that the LoginModule currently creates and adjusting the SASL Client callback handler accordingly.  For example, the org.apache.kafka.common.security.plain.PlainLoginModule class currently adds a String to the private credentials for the password and a String to the public credentials for the username; it would instead store an instance of a class that implements ExpiringCredential and includes an additional password() method, and the org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler class would be amended to pull the data from that object if it exists (rather than from the pair of String values, which it would fall back to if no ExpiringCredential existed).  Similar adjustments could be done for the SCRAM-related mechanisms as well.  The GSSAPI mechanism would be a bit different because we don't control all the code (we reuse an existing LoginModule implementation, for example); in such cases we would simply use our own implementation that delegates to existing one and that augments where necessary (e.g. in the commit() method of the LoginModule implementation).

Getting the client and server to agree on a credential expiration time may sometimes be a challenge.  It is easy with OAUTHBEARER because bearer tokens advertise their expiration times (either as part of the JWT or via a call to the token endpoint).  GSSAPI also is likely to be straightforward because Kerberos tickets have an expiration time.  What about PLAIN and the SCRAM-related SASL mechanisms?  These present challenges.  The SCRAM-related SASL mechanisms support SASL extensions, so we could leverage this to have the client send an agreed-upon name/value pair to the server indicating the credential lifetime it desires; the server could fail the authentication if the requested lifetime exceeds its maximum allowed value, but otherwise it would accept the lifetime that the client wants and proceed as it does today. The PLAIN SASL mechanism does not support extensions, so we would have to come up with another option (which we could also use for the SCRAM-related mechanisms if we wanted to be consistent).  We could add a new API call so that clients could ask servers for the lifetime they use, or we could extend the SaslHandshakeRequest/Response API call to include that information in the server's response – the client would then adopt that value.  Or we could just configure the values on the client and server and require the configurations to be consistent with each other (i.e. the client-side configuration value must not exceed the server-side configuration value).

Getting the client and server to agree on the credential lifetime is the challenge; discussing the options, documenting the solution, and implementing it would all be done as part of a KIP.

Compatibility, Deprecation, and Migration Plan

...