@Test
  public void testGetMetadataWithPrefetch() {
    SystemStreamPartition ssp = buildSSP(0);
    SystemStreamPartition otherSSP = buildSSP(1);
    SSPMetadataCache cache = buildSSPMetadataCache(ImmutableSet.of(ssp, otherSSP));

    // t = 10: first read, t = 11: first write
    when(clock.currentTimeMillis()).thenReturn(10L, 11L);
    when(systemAdmin.getSSPMetadata(ImmutableSet.of(ssp, otherSSP))).thenReturn(
        ImmutableMap.of(ssp, sspMetadata(1), otherSSP, sspMetadata(2)));
    assertEquals(sspMetadata(1), cache.getMetadata(ssp));
    verify(systemAdmin).getSSPMetadata(ImmutableSet.of(ssp, otherSSP));

    // stay within TTL: use cached data
    when(clock.currentTimeMillis()).thenReturn(11 + CACHE_TTL.toMillis());
    assertEquals(sspMetadata(1), cache.getMetadata(ssp));
    assertEquals(sspMetadata(2), cache.getMetadata(otherSSP));
    // still only one call to the admin from the initial fill
    verify(systemAdmin).getSSPMetadata(ImmutableSet.of(ssp, otherSSP));

    // now entries are stale
    when(clock.currentTimeMillis()).thenReturn(12 + CACHE_TTL.toMillis());
    when(systemAdmin.getSSPMetadata(ImmutableSet.of(ssp, otherSSP))).thenReturn(
        ImmutableMap.of(ssp, sspMetadata(10), otherSSP, sspMetadata(11)));
    // flip the order; prefetching should still be done correctly
    assertEquals(sspMetadata(11), cache.getMetadata(otherSSP));
    assertEquals(sspMetadata(10), cache.getMetadata(ssp));
    verify(systemAdmin, times(2)).getSSPMetadata(ImmutableSet.of(ssp, otherSSP));
  }
Given that there are sspsToPrefetch, getMetadata should call the admin (when necessary) to get the metadata for the
requested and "prefetch" SSPs. It should also cache the data.
samza-core/src/test/java/org/apache/samza/system/TestSSPMetadataCache.java