@Test
    public void testTokenReauthenticationOverSaslScram() throws Exception {
        SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
        TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Collections.singletonList("SCRAM-SHA-256"));

        // create jaas config for token auth
        Map<String, Object> options = new HashMap<>();
        String tokenId = "token1";
        String tokenHmac = "abcdefghijkl";
        options.put("username", tokenId); // tokenId
        options.put("password", tokenHmac); // token hmac
        options.put(ScramLoginModule.TOKEN_AUTH_CONFIG, "true"); // enable token authentication
        jaasConfig.createOrUpdateEntry(TestJaasConfig.LOGIN_CONTEXT_CLIENT, ScramLoginModule.class.getName(), options);

        // ensure re-authentication based on token expiry rather than a default value
        saslServerConfigs.put(BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_CONFIG, Long.MAX_VALUE);
        /*
         * create a token cache that adjusts the token expiration dynamically so that
         * the first time the expiry is read during authentication we use it to define a
         * session expiration time that we can then sleep through; then the second time
         * the value is read (during re-authentication) it will be in the future.
         */
        Function<Integer, Long> tokenLifetime = callNum -> 10 * callNum * CONNECTIONS_MAX_REAUTH_MS_VALUE;
        DelegationTokenCache tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames()) {
            int callNum = 0;

            @Override
            public TokenInformation token(String tokenId) {
                TokenInformation baseTokenInfo = super.token(tokenId);
                long thisLifetimeMs = System.currentTimeMillis() + tokenLifetime.apply(++callNum);
                return new TokenInformation(baseTokenInfo.tokenId(), baseTokenInfo.owner(),
                        baseTokenInfo.renewers(), baseTokenInfo.issueTimestamp(), thisLifetimeMs, thisLifetimeMs);
            }
        };
        server = createEchoServer(ListenerName.forSecurityProtocol(securityProtocol), securityProtocol, tokenCache);

        KafkaPrincipal owner = SecurityUtils.parseKafkaPrincipal("User:Owner");
        KafkaPrincipal renewer = SecurityUtils.parseKafkaPrincipal("User:Renewer1");
        TokenInformation tokenInfo = new TokenInformation(tokenId, owner, Collections.singleton(renewer),
                System.currentTimeMillis(), System.currentTimeMillis(), System.currentTimeMillis());
        server.tokenCache().addToken(tokenId, tokenInfo);
        updateTokenCredentialCache(tokenId, tokenHmac);
        // initial authentication must succeed
        createClientConnection(securityProtocol, "0");
        checkClientConnection("0");
        // ensure metrics are as expected before trying to re-authenticate
        server.verifyAuthenticationMetrics(1, 0);
        server.verifyReauthenticationMetrics(0, 0);
        /*
         * Now re-authenticate and ensure it succeeds. We have to sleep long enough so
         * that the current delegation token will be expired when the next write occurs;
         * this will trigger a re-authentication. Then the second time the delegation
         * token is read and transmitted to the server it will again have an expiration
         * date in the future.
         */
        delay(tokenLifetime.apply(1));
        checkClientConnection("0");
        server.verifyReauthenticationMetrics(1, 0);
    }
Tests SASL/SCRAM with username containing characters that need
to be encoded.
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java