@Override
    public byte[] evaluateResponse(byte[] responseBytes) throws SaslAuthenticationException {
        /*
         * Message format (from https://tools.ietf.org/html/rfc4616):
         *
         * message   = [authzid] UTF8NUL authcid UTF8NUL passwd
         * authcid   = 1*SAFE ; MUST accept up to 255 octets
         * authzid   = 1*SAFE ; MUST accept up to 255 octets
         * passwd    = 1*SAFE ; MUST accept up to 255 octets
         * UTF8NUL   = %x00 ; UTF-8 encoded NUL character
         *
         * SAFE      = UTF1 / UTF2 / UTF3 / UTF4
         *                ;; any UTF-8 encoded Unicode character except NUL
         */

        String response = new String(responseBytes, StandardCharsets.UTF_8);
        List<String> tokens = extractTokens(response);
        String authorizationIdFromClient = tokens.get(0);
        String username = tokens.get(1);
        String password = tokens.get(2);

        if (username.isEmpty()) {
            throw new SaslAuthenticationException("Authentication failed: username not specified");
        }
        if (password.isEmpty()) {
            throw new SaslAuthenticationException("Authentication failed: password not specified");
        }

        NameCallback nameCallback = new NameCallback("username", username);
        PlainAuthenticateCallback authenticateCallback = new PlainAuthenticateCallback(password.toCharArray());
        try {
            callbackHandler.handle(new Callback[]{nameCallback, authenticateCallback});
        } catch (Throwable e) {
            throw new SaslAuthenticationException("Authentication failed: credentials for user could not be verified", e);
        }
        if (!authenticateCallback.authenticated())
            throw new SaslAuthenticationException("Authentication failed: Invalid username or password");
        if (!authorizationIdFromClient.isEmpty() && !authorizationIdFromClient.equals(username))
            throw new SaslAuthenticationException("Authentication failed: Client requested an authorization id that is different from username");

        this.authorizationId = username;

        complete = true;
        return new byte[0];
    }
@throws SaslAuthenticationException if username/password combination is invalid or if the requested
        authorization id is not the same as username.
<p>
<b>Note:</b> This method may throw {@link SaslAuthenticationException} to provide custom error messages
to clients. But care should be taken to avoid including any information in the exception message that
should not be leaked to unauthenticated clients. It may be safer to throw {@link SaslException} in
some cases so that a standard error message is returned to clients.
</p>
clients/src/main/java/org/apache/kafka/common/security/plain/internals/PlainSaslServer.java