fix: Support for User Credentials

[stankiewicz]

I faced issue: Unknown credentials type UserCredentials
steps to reproduce:

gcloud auth application-default login
configure callback handler in properties
run kafka producer
Fix is to add support for UserCredentials and pull subject.

[sjvanrossum]

@stankiewicz I was looking into the auth handler last week and what I think this handler on the client should be doing imho is converting the application default credentials into IdTokenCredentials with Option.FORMAT_FULL and Option.INCLUDE_EMAIL (supported for ServiceAccountCredentials, ComputeEngineCredentials, ImpersonatedCredentials and UserCredentials) and sending the resulting token during the SASL exchange instead of wrapping the existing token with the boilerplate defined at the top of GcpLoginCallbackHandler and passing the actual token as the signature of the JWT. The current token format does not conform to a format defined by the standard and unnecessarily complicates the token validation on the server side in my opinion.

Example (requires changes in the server auth handler and the :authenticateConnection endpoint):

IdTokenCredentials.newBuilder()
    .setTargetAudience("managedkafka.googleapis.com")
    .setIdTokenProvider((IdTokenProvider) GoogleCredentials.getApplicationDefault())
    .setOptions(Arrays.asList(
        IdTokenProvider.Option.FORMAT_FULL,
        IdTokenProvider.Option.INCLUDE_EMAIL))
    .build()
    .refreshAccessToken()
    .getTokenValue();

[sjvanrossum]

@stankiewicz working with the existing token format, you can use the following snippet to generate the subject for any IdTokenProvider or an IdTokenProvider that doesn't already offer its account identifier as part of the credentials like UserCredentials (works):

import com.google.api.client.googleapis.auth.oauth2.GoogleIdToken;
import com.google.api.client.json.JsonFactory;
import com.google.api.client.json.gson.GsonFactory;
import com.google.auth.oauth2.AccessToken;
import com.google.auth.oauth2.IdTokenCredentials;
import com.google.auth.oauth2.IdTokenProvider;
import java.util.Arrays;

...

private static final JsonFactory JSON_FACTORY = new GsonFactory();

...

if (credentials instanceof IdTokenProvider) {
  subject =
      GoogleIdToken.parse(
              JSON_FACTORY,
              IdTokenCredentials.newBuilder()
                  .setTargetAudience("https://managedkafka.googleapis.com")
                  .setOptions(
                      Arrays.asList(
                          IdTokenProvider.Option.FORMAT_FULL,
                          IdTokenProvider.Option.INCLUDE_EMAIL))
                  .setIdTokenProvider((IdTokenProvider) credentials)
                  .build()
                  .refreshAccessToken()
                  .getTokenValue())
          .getPayload()
          .getEmail();
}

[acocuzzo]

I did not have the permissions to modify this PR so I created a new one with @sjvanrossum suggestions: #31
@stankiewicz PTAL a look

[acocuzzo]

Closing as alternative fix was merged.