@Test
    public void testLeaderMetadataInconsistentWithBrokerMetadata() {
        // Tests a reordering scenario which can lead to inconsistent leader state.
        // A partition initially has one broker offline. That broker comes online and
        // is elected leader. The client sees these two events in the opposite order.

        TopicPartition tp = new TopicPartition("topic", 0);

        Node node0 = new Node(0, "localhost", 9092);
        Node node1 = new Node(1, "localhost", 9093);
        Node node2 = new Node(2, "localhost", 9094);

        // The first metadata received by broker (epoch=10)
        MetadataResponsePartition firstPartitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setErrorCode(Errors.NONE.code())
                .setLeaderEpoch(10)
                .setLeaderId(0)
                .setReplicaNodes(Arrays.asList(0, 1, 2))
                .setIsrNodes(Arrays.asList(0, 1, 2))
                .setOfflineReplicas(Collections.emptyList());

        // The second metadata received has stale metadata (epoch=8)
        MetadataResponsePartition secondPartitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setErrorCode(Errors.NONE.code())
                .setLeaderEpoch(8)
                .setLeaderId(1)
                .setReplicaNodes(Arrays.asList(0, 1, 2))
                .setIsrNodes(Arrays.asList(1, 2))
                .setOfflineReplicas(Collections.singletonList(0));

        metadata.updateWithCurrentRequestVersion(new MetadataResponse(new MetadataResponseData()
                        .setTopics(buildTopicCollection(tp.topic(), firstPartitionMetadata))
                        .setBrokers(buildBrokerCollection(Arrays.asList(node0, node1, node2))),
                        ApiKeys.METADATA.latestVersion()),
                false, 10L);

        metadata.updateWithCurrentRequestVersion(new MetadataResponse(new MetadataResponseData()
                        .setTopics(buildTopicCollection(tp.topic(), secondPartitionMetadata))
                        .setBrokers(buildBrokerCollection(Arrays.asList(node1, node2))),
                        ApiKeys.METADATA.latestVersion()),
                false, 20L);

        assertNull(metadata.fetch().leaderFor(tp));
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));
        assertFalse(metadata.currentLeader(tp).leader.isPresent());
    }
Prior to Kafka version 2.4 (which coincides with Metadata version 9), the broker does not propagate leader epoch
information accurately while a reassignment is in progress, so we cannot rely on it. This is explained in more
detail in MetadataResponse's constructor.
clients/src/test/java/org/apache/kafka/clients/MetadataTest.java