private void maybeRefreshMetadata(SystemStreamPartition requestedSSP) {
    synchronized (this.metadataRefreshLock) {
      Instant refreshRequestedAt = Instant.ofEpochMilli(this.clock.currentTimeMillis());
      if (shouldRefresh(requestedSSP, refreshRequestedAt)) {
        String system = requestedSSP.getSystem();
        Set<SystemStreamPartition> sspsToFetchFor = new HashSet<>();
        sspsToFetchFor.add(requestedSSP);
        for (SystemStreamPartition sspToPrefetch : this.sspsToPrefetch) {
          if (system.equals(sspToPrefetch.getSystem()) && shouldRefresh(sspToPrefetch, refreshRequestedAt)) {
            sspsToFetchFor.add(sspToPrefetch);
          }
        }
        SystemAdmin systemAdmin = this.systemAdmins.getSystemAdmin(system);
        Map<SystemStreamPartition, SystemStreamMetadata.SystemStreamPartitionMetadata> fetchedMetadata =
            systemAdmin.getSSPMetadata(sspsToFetchFor);
        Instant updatedAt = Instant.ofEpochMilli(this.clock.currentTimeMillis());
        // we want to add an entry even if there was no metadata, so iterate over sspsToFetchFor
        sspsToFetchFor.forEach(ssp -> this.cache.put(ssp, new CacheEntry(fetchedMetadata.get(ssp), updatedAt)));
      }
    }
  }
Gets the metadata for an SSP. This will return a cached value if it is fresh enough. Otherwise, it will fetch the
metadata from a source-of-truth.
If the metadata for the SSP needs to be fetched, then this will also prefetch and cache the metadata for any stale
{@link #sspsToPrefetch} that can be included in the same fetch call (e.g. same system).

@param ssp SSP for which to get metadata
@return metadata for the SSP; null if the source-of-truth returned no metadata
@throws RuntimeException if there was an error in fetching metadata
samza-core/src/main/scala/org/apache/samza/system/SSPMetadataCache.java