private SaslServer createSaslKerberosServer(final AuthenticateCallbackHandler saslServerCallbackHandler, final Map<String, ?> configs, Subject subject) throws IOException {
        // server is using a JAAS-authenticated subject: determine service principal name and hostname from kafka server's subject.
        final String servicePrincipal = SaslClientAuthenticator.firstPrincipal(subject);
        KerberosName kerberosName;
        try {
            kerberosName = KerberosName.parse(servicePrincipal);
        } catch (IllegalArgumentException e) {
            throw new KafkaException("Principal has name with unexpected format " + servicePrincipal);
        }
        final String servicePrincipalName = kerberosName.serviceName();
        final String serviceHostname = kerberosName.hostName();

        LOG.debug("Creating SaslServer for {} with mechanism {}", kerberosName, saslMechanism);

        try {
            return Subject.doAs(subject, (PrivilegedExceptionAction<SaslServer>) () ->
                    Sasl.createSaslServer(saslMechanism, servicePrincipalName, serviceHostname, configs, saslServerCallbackHandler));
        } catch (PrivilegedActionException e) {
            throw new SaslException("Kafka Server failed to create a SaslServer to interact with a client during session authentication", 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