@Override
    public byte[] evaluateChallenge(byte[] challenge) throws SaslException {
        try {
            OAuthBearerTokenCallback callback = new OAuthBearerTokenCallback();
            switch (state) {
                case SEND_CLIENT_FIRST_MESSAGE:
                    if (challenge != null && challenge.length != 0)
                        throw new SaslException("Expected empty challenge");
                    callbackHandler().handle(new Callback[] {callback});
                    SaslExtensions extensions = retrieveCustomExtensions();

                    setState(State.RECEIVE_SERVER_FIRST_MESSAGE);

                    return new OAuthBearerClientInitialResponse(callback.token().value(), extensions).toBytes();
                case RECEIVE_SERVER_FIRST_MESSAGE:
                    if (challenge != null && challenge.length != 0) {
                        String jsonErrorResponse = new String(challenge, StandardCharsets.UTF_8);
                        if (log.isDebugEnabled())
                            log.debug("Sending %%x01 response to server after receiving an error: {}",
                                    jsonErrorResponse);
                        setState(State.RECEIVE_SERVER_MESSAGE_AFTER_FAILURE);
                        return new byte[] {BYTE_CONTROL_A};
                    }
                    callbackHandler().handle(new Callback[] {callback});
                    if (log.isDebugEnabled())
                        log.debug("Successfully authenticated as {}", callback.token().principalName());
                    setState(State.COMPLETE);
                    return null;
                default:
                    throw new IllegalSaslStateException("Unexpected challenge in Sasl client state " + state);
            }
        } catch (SaslException e) {
            setState(State.FAILED);
            throw e;
        } catch (IOException | UnsupportedCallbackException e) {
            setState(State.FAILED);
            throw new SaslException(e.getMessage(), e);
        }
    }
{@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