public OptionalInt leaderEpochFor(TopicPartition tp) {
        PartitionMetadata partitionMetadata = metadataByPartition.get(tp);
        if (partitionMetadata == null || !partitionMetadata.leaderEpoch.isPresent()) {
            return OptionalInt.empty();
        } else {
            return OptionalInt.of(partitionMetadata.leaderEpoch.get());
        }
    }
@Test
    public void testLeaderEpochFor() {
        // Setup partition 0 with a leader-epoch of 10.
        TopicPartition topicPartition1 = new TopicPartition("topic", 0);
        MetadataResponse.PartitionMetadata partitionMetadata1 = new MetadataResponse.PartitionMetadata(
            Errors.NONE,
            topicPartition1,
            Optional.of(5),
            Optional.of(10),
            Arrays.asList(5, 6, 7),
            Arrays.asList(5, 6, 7),
            Collections.emptyList());

        // Setup partition 1 with an unknown leader epoch.
        TopicPartition topicPartition2 = new TopicPartition("topic", 1);
        MetadataResponse.PartitionMetadata partitionMetadata2 = new MetadataResponse.PartitionMetadata(
            Errors.NONE,
            topicPartition2,
            Optional.of(5),
            Optional.empty(),
            Arrays.asList(5, 6, 7),
            Arrays.asList(5, 6, 7),
            Collections.emptyList());

        Map<Integer, Node> nodesById = new HashMap<>();
        nodesById.put(5, new Node(5, "localhost", 2077));
        nodesById.put(6, new Node(6, "localhost", 2078));
        nodesById.put(7, new Node(7, "localhost", 2079));

        MetadataSnapshot cache = new MetadataSnapshot("clusterId",
            nodesById,
            Arrays.asList(partitionMetadata1, partitionMetadata2),
            Collections.emptySet(),
            Collections.emptySet(),
            Collections.emptySet(),
            null,
            Collections.emptyMap());

        assertEquals(OptionalInt.of(10), cache.leaderEpochFor(topicPartition1));

        assertEquals(OptionalInt.empty(), cache.leaderEpochFor(topicPartition2));

        assertEquals(OptionalInt.empty(), cache.leaderEpochFor(new TopicPartition("topic_missing", 0)));
    }