feat: Add segmentio/kafka-go Mechanism implementation for SASL/Plain with GCP OAuth Tokens.

[benjamin-maynard]

Internal Note: This depends on a rollout of cl/712634153.

This package provides an implementation of the sasl.Mechanism interface from segmentio/kafka-go that handles authentication to Google Managed Kafka using OAuth Tokens from Application Default Credentials.

It allows you to use Authorization Tokens with SASL/Plain in segmentio/kafka-go, without requiring OAuthBearer support in the library.

This is useful for GMK customers that use this library for Kafka - who cannot use OAuthBearer authentication due to lack of support.

[davidtorres]

I'd remove this file

[benjamin-maynard]

Happy to remove - would you mind expanding on why it should be removed?

[davidtorres]

how does segmentio io knows it has to refresh the token?

[benjamin-maynard]

It doesn't know it has to refresh the token - but it calls Start¹ for every new Kafka broker connection and uses the returned token for authenticating the connection.

The TokenSource in the Mechanism always returns a valid token at the point it is called, and the token is only validated on the broker at the point of connection establishment, so as long as it is valid at the point the connection is established, then we no longer care about expiry for that connection.

[1] Each time the Dialer connects to a Kafka Broker, it calls Dialer.authenticateSASL if there is a non nil SASLMechanism, which in turn calls the Start receiver function on the Mechanism. The encoded username and password (token) is then used for authentication to the broker.