private void createSaslServer(String mechanism) throws IOException {
        this.saslMechanism = mechanism;
        Subject subject = subjects.get(mechanism);
        final AuthenticateCallbackHandler callbackHandler = callbackHandlers.get(mechanism);
        if (mechanism.equals(SaslConfigs.GSSAPI_MECHANISM)) {
            saslServer = createSaslKerberosServer(callbackHandler, configs, subject);
        } else {
            try {
                saslServer = Subject.doAs(subject, (PrivilegedExceptionAction<SaslServer>) () ->
                    Sasl.createSaslServer(saslMechanism, "kafka", serverAddress().getHostName(), configs, callbackHandler));
                if (saslServer == null) {
                    throw new SaslException("Kafka Server failed to create a SaslServer to interact with a client during session authentication with server mechanism " + saslMechanism);
                }
            } catch (PrivilegedActionException e) {
                throw new SaslException("Kafka Server failed to create a SaslServer to interact with a client during session authentication with server mechanism " + saslMechanism, e.getCause());
            }
        }
    }
The internal state transitions for initial authentication of a channel on the
server side are declared in order, starting with {@link #INITIAL_REQUEST} and
ending in either {@link #COMPLETE} or {@link #FAILED}.
<p>
Re-authentication of a channel on the server side starts with the state
{@link #REAUTH_PROCESS_HANDSHAKE}. It may then flow to
{@link #REAUTH_BAD_MECHANISM} before a transition to {@link #FAILED} if
re-authentication is attempted with a mechanism different than the original
one; otherwise it joins the authentication flow at the {@link #AUTHENTICATE}
state and likewise ends at either {@link #COMPLETE} or {@link #FAILED}.
clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java