@Override
        public SaslClient createSaslClient(String[] mechanisms, String authorizationId, String protocol,
                String serverName, Map<String, ?> props, CallbackHandler callbackHandler) {
            String[] mechanismNamesCompatibleWithPolicy = getMechanismNames(props);
            for (String mechanism : mechanisms) {
                for (String name : mechanismNamesCompatibleWithPolicy) {
                    if (name.equals(mechanism)) {
                        if (!(Objects.requireNonNull(callbackHandler) instanceof AuthenticateCallbackHandler))
                            throw new IllegalArgumentException(String.format(
                                    "Callback handler must be castable to %s: %s",
                                    AuthenticateCallbackHandler.class.getName(), callbackHandler.getClass().getName()));
                        return new OAuthBearerSaslClient((AuthenticateCallbackHandler) callbackHandler);
                    }
                }
            }
            return null;
        }
{@code SaslClient} implementation for SASL/OAUTHBEARER in Kafka. This
implementation requires an instance of {@code AuthenticateCallbackHandler}
that can handle an instance of {@link OAuthBearerTokenCallback} and return
the {@link OAuthBearerToken} generated by the {@code login()} event on the
{@code LoginContext}. Said handler can also optionally handle an instance of {@link SaslExtensionsCallback}
to return any extensions generated by the {@code login()} event on the {@code LoginContext}.

@see <a href="https://tools.ietf.org/html/rfc6750#section-2.1">RFC 6750,
     Section 2.1</a>
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/OAuthBearerSaslClient.java