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;
    }
@Test
    public void testRefreshWithMinPeriodIntrusion() throws Exception {
        int numExpectedRefreshes = 1;
        boolean clientReloginAllowedBeforeLogout = true;
        Subject subject = new Subject();
        final LoginContext mockLoginContext = mock(LoginContext.class);
        when(mockLoginContext.getSubject()).thenReturn(subject);

        MockTime mockTime = new MockTime();
        long startMs = mockTime.milliseconds();
        /*
         * Identify the lifetime of each expiring credential
         */
        long lifetimeMinutes = 10L;
        /*
         * Identify the point at which refresh will occur in that lifetime
         */
        long refreshEveryMinutes = 8L;
        /*
         * Set an absolute last refresh time that will cause the login thread to exit
         * after a certain number of re-logins (by adding an extra half of a refresh
         * interval).
         */
        long absoluteLastRefreshMs = startMs + (1 + numExpectedRefreshes) * 1000 * 60 * refreshEveryMinutes
                - 1000 * 60 * refreshEveryMinutes / 2;

        /*
         * Identify a minimum period that will cause the refresh time to be delayed a
         * bit.
         */
        int bufferIntrusionSeconds = 1;
        short minPeriodSeconds = (short) (refreshEveryMinutes * 60 + bufferIntrusionSeconds);
        short bufferSeconds = (short) 0;

        /*
         * Define some listeners so we can keep track of who gets done and when. All
         * added listeners should end up done except the last, extra one, which should
         * not.
         */
        MockScheduler mockScheduler = new MockScheduler(mockTime);
        List<KafkaFutureImpl<Long>> waiters = addWaiters(mockScheduler,
                1000 * (60 * refreshEveryMinutes + bufferIntrusionSeconds), numExpectedRefreshes + 1);

        // Create the ExpiringCredentialRefreshingLogin instance under test
        TestLoginContextFactory testLoginContextFactory = new TestLoginContextFactory();
        TestExpiringCredentialRefreshingLogin testExpiringCredentialRefreshingLogin = new TestExpiringCredentialRefreshingLogin(
                refreshConfigThatPerformsReloginEveryGivenPercentageOfLifetime(
                        1.0 * refreshEveryMinutes / lifetimeMinutes, minPeriodSeconds, bufferSeconds,
                        clientReloginAllowedBeforeLogout),
                testLoginContextFactory, mockTime, 1000 * 60 * lifetimeMinutes, absoluteLastRefreshMs,
                clientReloginAllowedBeforeLogout);
        testLoginContextFactory.configure(mockLoginContext, testExpiringCredentialRefreshingLogin);

        /*
         * Perform the login, wait up to a certain amount of time for the refresher
         * thread to exit, and make sure the correct calls happened at the correct times
         */
        long expectedFinalMs = startMs
                + numExpectedRefreshes * 1000 * (60 * refreshEveryMinutes + bufferIntrusionSeconds);
        assertFalse(testLoginContextFactory.refresherThreadStartedFuture().isDone());
        assertFalse(testLoginContextFactory.refresherThreadDoneFuture().isDone());
        testExpiringCredentialRefreshingLogin.login();
        assertTrue(testLoginContextFactory.refresherThreadStartedFuture().isDone());
        testLoginContextFactory.refresherThreadDoneFuture().get(1L, TimeUnit.SECONDS);
        assertEquals(expectedFinalMs, mockTime.milliseconds());
        for (int i = 0; i < numExpectedRefreshes; ++i) {
            KafkaFutureImpl<Long> waiter = waiters.get(i);
            assertTrue(waiter.isDone());
            assertEquals((i + 1) * 1000 * (60 * refreshEveryMinutes + bufferIntrusionSeconds),
                    waiter.get() - startMs);
        }
        assertFalse(waiters.get(numExpectedRefreshes).isDone());

        InOrder inOrder = inOrder(mockLoginContext);
        inOrder.verify(mockLoginContext).login();
        for (int i = 0; i < numExpectedRefreshes; ++i) {
            inOrder.verify(mockLoginContext).login();
            inOrder.verify(mockLoginContext).logout();
        }
    }