public LoginContext login() throws LoginException {
        LoginContext tmpLoginContext = loginContextFactory.createLoginContext(this);
        tmpLoginContext.login();
        log.info("Successfully logged in.");
        loginContext = tmpLoginContext;
        subject = loginContext.getSubject();
        expiringCredential = expiringCredential();
        hasExpiringCredential = expiringCredential != null;
        if (!hasExpiringCredential) {
            // do not bother with re-logins.
            log.debug("No Expiring Credential");
            principalName = null;
            refresherThread = null;
            return loginContext;
        }

        principalName = expiringCredential.principalName();

        // Check for a clock skew problem
        long expireTimeMs = expiringCredential.expireTimeMs();
        long nowMs = currentMs();
        if (nowMs > expireTimeMs) {
            log.error(
                    "[Principal={}]: Current clock: {} is later than expiry {}. This may indicate a clock skew problem."
                            + " Check that this host's and remote host's clocks are in sync. Not starting refresh thread."
                            + " This process is likely unable to authenticate SASL connections (for example, it is unlikely"
                            + " to be able to authenticate a connection with a Kafka Broker).",
                    principalLogText(), new Date(nowMs), new Date(expireTimeMs));
            return loginContext;
        }

        if (log.isDebugEnabled())
            log.debug("[Principal={}]: It is an expiring credential", principalLogText());

        /*
         * Re-login periodically. How often is determined by the expiration date of the
         * credential and refresh-related configuration values.
         */
        refresherThread = KafkaThread.daemon(String.format("kafka-expiring-relogin-thread-%s", principalName),
                new Refresher());
        refresherThread.start();
        loginContextFactory.refresherThreadStarted();
        return loginContext;
    }
Performs login for each login module specified for the login context of this
instance and starts the thread used to periodically re-login.
<p>
The synchronized keyword is not necessary because an implementation of
{@link Login} will delegate to this code (e.g. OAuthBearerRefreshingLogin),
and the {@code login()} method on the delegating class will itself be
synchronized if necessary.
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/expiring/ExpiringCredentialRefreshingLogin.java