private Optional<MetadataResponse.PartitionMetadata> updateLatestMetadata(
            MetadataResponse.PartitionMetadata partitionMetadata,
            boolean hasReliableLeaderEpoch,
            Uuid topicId,
            Uuid oldTopicId) {
        TopicPartition tp = partitionMetadata.topicPartition;
        if (hasReliableLeaderEpoch && partitionMetadata.leaderEpoch.isPresent()) {
            int newEpoch = partitionMetadata.leaderEpoch.get();
            Integer currentEpoch = lastSeenLeaderEpochs.get(tp);
            if (currentEpoch == null) {
                // We have no previous info, so we can just insert the new epoch info
                log.debug("Setting the last seen epoch of partition {} to {} since the last known epoch was undefined.",
                        tp, newEpoch);
                lastSeenLeaderEpochs.put(tp, newEpoch);
                this.equivalentResponseCount = 0;
                return Optional.of(partitionMetadata);
            } else if (topicId != null && !topicId.equals(oldTopicId)) {
                // If the new topic ID is valid and different from the last seen topic ID, update the metadata.
                // Between the time that a topic is deleted and re-created, the client may lose track of the
                // corresponding topicId (i.e. `oldTopicId` will be null). In this case, when we discover the new
                // topicId, we allow the corresponding leader epoch to override the last seen value.
                log.info("Resetting the last seen epoch of partition {} to {} since the associated topicId changed from {} to {}",
                        tp, newEpoch, oldTopicId, topicId);
                lastSeenLeaderEpochs.put(tp, newEpoch);
                this.equivalentResponseCount = 0;
                return Optional.of(partitionMetadata);
            } else if (newEpoch >= currentEpoch) {
                // If the received leader epoch is at least the same as the previous one, update the metadata
                log.debug("Updating last seen epoch for partition {} from {} to epoch {} from new metadata", tp, currentEpoch, newEpoch);
                lastSeenLeaderEpochs.put(tp, newEpoch);
                if (newEpoch > currentEpoch) {
                    this.equivalentResponseCount = 0;
                }
                return Optional.of(partitionMetadata);
            } else {
                // Otherwise ignore the new metadata and use the previously cached info
                log.debug("Got metadata for an older epoch {} (current is {}) for partition {}, not updating", newEpoch, currentEpoch, tp);
                return metadataSnapshot.partitionMetadata(tp);
            }
        } else {
            // Handle old cluster formats as well as error responses where leader and epoch are missing
            lastSeenLeaderEpochs.remove(tp);
            this.equivalentResponseCount = 0;
            return Optional.of(partitionMetadata.withoutLeaderEpoch());
        }
    }
Compute the latest partition metadata to cache given ordering by leader epochs (if both
available and reliable) and whether the topic ID changed.
clients/src/main/java/org/apache/kafka/clients/Metadata.java