@Test
    public void testRepeatedValidSaslPlainOverSsl() throws Exception {
        String node = "0";
        SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
        configureMechanisms("PLAIN", Collections.singletonList("PLAIN"));
        /*
         * Make sure 85% of this value is at least 1 second otherwise it is possible for
         * the client to start re-authenticating but the server does not start due to
         * the 1-second minimum. If this happens the SASL HANDSHAKE request that was
         * injected to start re-authentication will be echoed back to the client instead
         * of the data that the client explicitly sent, and then the client will not
         * recognize that data and will throw an assertion error.
         */
        saslServerConfigs.put(BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_CONFIG,
                Double.valueOf(1.1 * 1000L / 0.85).longValue());

        server = createEchoServer(securityProtocol);
        createClientConnection(securityProtocol, node);
        checkClientConnection(node);
        server.verifyAuthenticationMetrics(1, 0);
        server.verifyReauthenticationMetrics(0, 0);
        double successfulReauthentications = 0;
        int desiredNumReauthentications = 5;
        long startMs = Time.SYSTEM.milliseconds();
        long timeoutMs = startMs + 1000 * 15; // stop after 15 seconds
        while (successfulReauthentications < desiredNumReauthentications
                && Time.SYSTEM.milliseconds() < timeoutMs) {
            checkClientConnection(node);
            successfulReauthentications = server.metricValue("successful-reauthentication-total");
        }
        server.verifyReauthenticationMetrics(desiredNumReauthentications, 0);
    }
Tests good path SASL/PLAIN client and server channels using SSL transport layer.
Repeatedly tests successful re-authentication over several seconds.
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java