@Test
    public void testIgnoreLeaderEpochInOlderMetadataResponse() {
        TopicPartition tp = new TopicPartition("topic", 0);

        MetadataResponsePartition partitionMetadata = new MetadataResponsePartition()
                .setPartitionIndex(tp.partition())
                .setLeaderId(5)
                .setLeaderEpoch(10)
                .setReplicaNodes(Arrays.asList(1, 2, 3))
                .setIsrNodes(Arrays.asList(1, 2, 3))
                .setOfflineReplicas(Collections.emptyList())
                .setErrorCode(Errors.NONE.code());

        MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
                .setName(tp.topic())
                .setErrorCode(Errors.NONE.code())
                .setPartitions(Collections.singletonList(partitionMetadata))
                .setIsInternal(false);

        MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
        topics.add(topicMetadata);

        MetadataResponseData data = new MetadataResponseData()
                .setClusterId("clusterId")
                .setControllerId(0)
                .setTopics(topics)
                .setBrokers(new MetadataResponseBrokerCollection());

        for (short version = ApiKeys.METADATA.oldestVersion(); version < 9; version++) {
            ByteBuffer buffer = MessageUtil.toByteBuffer(data, version);
            MetadataResponse response = MetadataResponse.parse(buffer, version);
            assertFalse(response.hasReliableLeaderEpochs());
            metadata.updateWithCurrentRequestVersion(response, false, 100);
            assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
            MetadataResponse.PartitionMetadata responseMetadata = this.metadata.partitionMetadataIfCurrent(tp).get();
            assertEquals(Optional.empty(), responseMetadata.leaderEpoch);
        }

        for (short version = 9; version <= ApiKeys.METADATA.latestVersion(); version++) {
            ByteBuffer buffer = MessageUtil.toByteBuffer(data, version);
            MetadataResponse response = MetadataResponse.parse(buffer, version);
            assertTrue(response.hasReliableLeaderEpochs());
            metadata.updateWithCurrentRequestVersion(response, false, 100);
            assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
            MetadataResponse.PartitionMetadata responseMetadata = metadata.partitionMetadataIfCurrent(tp).get();
            assertEquals(Optional.of(10), responseMetadata.leaderEpoch);
        }
    }
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