public synchronized void update(int requestVersion, MetadataResponse response, boolean isPartialUpdate, long nowMs) {
        Objects.requireNonNull(response, "Metadata response cannot be null");
        if (isClosed())
            throw new IllegalStateException("Update requested after metadata close");

        this.needPartialUpdate = requestVersion < this.requestVersion;
        this.lastRefreshMs = nowMs;
        this.attempts = 0;
        this.updateVersion += 1;
        if (!isPartialUpdate) {
            this.needFullUpdate = false;
            this.lastSuccessfulRefreshMs = nowMs;
        }
        // If we subsequently find that the metadata response is not equivalent to the metadata already known,
        // this count is reset to 0 in updateLatestMetadata()
        this.equivalentResponseCount++;

        String previousClusterId = metadataSnapshot.clusterResource().clusterId();

        this.metadataSnapshot = handleMetadataResponse(response, isPartialUpdate, nowMs);

        Cluster cluster = metadataSnapshot.cluster();
        maybeSetMetadataError(cluster);

        this.lastSeenLeaderEpochs.keySet().removeIf(tp -> !retainTopic(tp.topic(), false, nowMs));

        String newClusterId = metadataSnapshot.clusterResource().clusterId();
        if (!Objects.equals(previousClusterId, newClusterId)) {
            log.info("Cluster ID: {}", newClusterId);
        }
        clusterResourceListeners.onUpdate(metadataSnapshot.clusterResource());

        log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.metadataSnapshot);
    }
Updates the cluster metadata. If topic expiry is enabled, expiry time
is set for topics if required and expired topics are removed from the metadata.

@param requestVersion The request version corresponding to the update response, as provided by
    {@link #newMetadataRequestAndVersion(long)}.
@param response metadata response received from the broker
@param isPartialUpdate whether the metadata request was for a subset of the active topics
@param nowMs current time in milliseconds
clients/src/main/java/org/apache/kafka/clients/Metadata.java