@Test
  public void testGetMetadataNoSSPsToPrefetch() {
    SystemStreamPartition ssp = buildSSP(0);
    SSPMetadataCache cache = buildSSPMetadataCache(ImmutableSet.of());

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

    // stay within TTL: use cached data
    when(clock.currentTimeMillis()).thenReturn(11 + CACHE_TTL.toMillis());
    assertEquals(sspMetadata(1), cache.getMetadata(ssp));

    // now entry is stale
    when(clock.currentTimeMillis()).thenReturn(12 + CACHE_TTL.toMillis());
    when(systemAdmin.getSSPMetadata(ImmutableSet.of(ssp))).thenReturn(ImmutableMap.of(ssp, sspMetadata(10)));
    assertEquals(sspMetadata(10), cache.getMetadata(ssp));
    verify(systemAdmin, times(2)).getSSPMetadata(ImmutableSet.of(ssp));
  }
Given that there are no sspsToPrefetch, getMetadata should still fetch and cache metadata for a requested SSP.
samza-core/src/test/java/org/apache/samza/system/TestSSPMetadataCache.java