@Test
    public void testVersionLogic() {
        for (short version : LEADER_AND_ISR.allVersions()) {
            List<LeaderAndIsrPartitionState> partitionStates = asList(
                new LeaderAndIsrPartitionState()
                    .setTopicName("topic0")
                    .setPartitionIndex(0)
                    .setControllerEpoch(2)
                    .setLeader(0)
                    .setLeaderEpoch(10)
                    .setIsr(asList(0, 1))
                    .setPartitionEpoch(10)
                    .setReplicas(asList(0, 1, 2))
                    .setAddingReplicas(Collections.singletonList(3))
                    .setRemovingReplicas(Collections.singletonList(2)),
                new LeaderAndIsrPartitionState()
                    .setTopicName("topic0")
                    .setPartitionIndex(1)
                    .setControllerEpoch(2)
                    .setLeader(1)
                    .setLeaderEpoch(11)
                    .setIsr(asList(1, 2, 3))
                    .setPartitionEpoch(11)
                    .setReplicas(asList(1, 2, 3))
                    .setAddingReplicas(emptyList())
                    .setRemovingReplicas(emptyList()),
                new LeaderAndIsrPartitionState()
                    .setTopicName("topic1")
                    .setPartitionIndex(0)
                    .setControllerEpoch(2)
                    .setLeader(2)
                    .setLeaderEpoch(11)
                    .setIsr(asList(2, 3, 4))
                    .setPartitionEpoch(11)
                    .setReplicas(asList(2, 3, 4))
                    .setAddingReplicas(emptyList())
                    .setRemovingReplicas(emptyList())
            );

            List<Node> liveNodes = asList(
                new Node(0, "host0", 9090),
                new Node(1, "host1", 9091)
            );

            Map<String, Uuid> topicIds = new HashMap<>();
            topicIds.put("topic0", Uuid.randomUuid());
            topicIds.put("topic1", Uuid.randomUuid());

            LeaderAndIsrRequest request = new LeaderAndIsrRequest.Builder(version, 1, 2, 3, partitionStates,
                topicIds, liveNodes).build();

            List<LeaderAndIsrLiveLeader> liveLeaders = liveNodes.stream().map(n -> new LeaderAndIsrLiveLeader()
                .setBrokerId(n.id())
                .setHostName(n.host())
                .setPort(n.port())).collect(Collectors.toList());
            assertEquals(new HashSet<>(partitionStates), iterableToSet(request.partitionStates()));
            assertEquals(liveLeaders, request.liveLeaders());
            assertEquals(1, request.controllerId());
            assertEquals(2, request.controllerEpoch());
            assertEquals(3, request.brokerEpoch());

            ByteBuffer byteBuffer = request.serialize();
            LeaderAndIsrRequest deserializedRequest = new LeaderAndIsrRequest(new LeaderAndIsrRequestData(
                new ByteBufferAccessor(byteBuffer), version), version);

            // Adding/removing replicas is only supported from version 3, so the deserialized request won't have
            // them for earlier versions.
            if (version < 3) {
                partitionStates.get(0)
                    .setAddingReplicas(emptyList())
                    .setRemovingReplicas(emptyList());
            }

            // Prior to version 2, there were no TopicStates, so a map of Topic Ids from a list of
            // TopicStates is an empty map.
            if (version < 2) {
                topicIds = new HashMap<>();
            }

            //  In versions 2-4 there are TopicStates, but no topicIds, so deserialized requests will have
            //  Zero Uuids in place.
            if (version > 1 && version < 5) {
                topicIds.put("topic0", Uuid.ZERO_UUID);
                topicIds.put("topic1", Uuid.ZERO_UUID);
            }

            assertEquals(new HashSet<>(partitionStates), iterableToSet(deserializedRequest.partitionStates()));
            assertEquals(topicIds, deserializedRequest.topicIds());
            assertEquals(liveLeaders, deserializedRequest.liveLeaders());
            assertEquals(1, request.controllerId());
            assertEquals(2, request.controllerEpoch());
            assertEquals(3, request.brokerEpoch());
        }
    }
Verifies the logic we have in LeaderAndIsrRequest to present a unified interface across the various versions
works correctly. For example, `LeaderAndIsrPartitionState.topicName` is not serialized/deserialized in
recent versions, but we set it manually so that we can always present the ungrouped partition states
independently of the version.
clients/src/test/java/org/apache/kafka/common/requests/LeaderAndIsrRequestTest.java