@Test
    public void testCorrelationId() {
        SaslClientAuthenticator authenticator = new SaslClientAuthenticator(
              Collections.emptyMap(),
              null,
              "node",
              null,
              null,
              null,
              "plain",
              false,
              null,
              null,
            new LogContext()
        ) {
            @Override
            SaslClient createSaslClient() {
                return null;
            }
        };
        int count = (SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID - SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID) * 2;
        Set<Integer> ids = IntStream.range(0, count)
            .mapToObj(i -> authenticator.nextCorrelationId())
            .collect(Collectors.toSet());
        assertEquals(SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID - SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID + 1, ids.size());
        ids.forEach(id -> {
            assertTrue(id >= SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID);
            assertTrue(SaslClientAuthenticator.isReserved(id));
        });
    }
Re-authentication must fail if principal changes
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java