@Test
    public void testConvertListOffsetResponseToSaslHandshakeResponse() {
        ListOffsetsResponseData data = new ListOffsetsResponseData()
                .setThrottleTimeMs(0)
                .setTopics(Collections.singletonList(new ListOffsetsTopicResponse()
                        .setName("topic")
                        .setPartitions(Collections.singletonList(new ListOffsetsPartitionResponse()
                                .setErrorCode(Errors.NONE.code())
                                .setLeaderEpoch(ListOffsetsResponse.UNKNOWN_EPOCH)
                                .setPartitionIndex(0)
                                .setOffset(0)
                                .setTimestamp(0)))));
        ListOffsetsResponse response = new ListOffsetsResponse(data);
        ByteBuffer buffer = RequestTestUtils.serializeResponseWithHeader(response, LIST_OFFSETS.latestVersion(), 0);
        final RequestHeader header0 = new RequestHeader(LIST_OFFSETS, LIST_OFFSETS.latestVersion(), "id", SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID);
        assertThrows(SchemaException.class, () -> NetworkClient.parseResponse(buffer.duplicate(), header0));
        final RequestHeader header1 = new RequestHeader(LIST_OFFSETS, LIST_OFFSETS.latestVersion(), "id", 1);
        assertThrows(IllegalStateException.class, () -> NetworkClient.parseResponse(buffer.duplicate(), header1));
    }
Re-authentication must fail if principal changes
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java