@Test
  public void testGetMetadataMultipleSystemsForPrefetch() {
    // add one more extended system admin so we can have two of them for this test
    SystemAdmin otherSystemAdmin = mock(SystemAdmin.class);
    String otherSystem = "otherSystem";
    when(systemAdmins.getSystemAdmin(otherSystem)).thenReturn(otherSystemAdmin);
    SystemStreamPartition ssp = buildSSP(0);
    // different system should not get prefetched
    SystemStreamPartition sspOtherSystem = new SystemStreamPartition(otherSystem, "otherStream", new Partition(1));
    SSPMetadataCache cache = buildSSPMetadataCache(ImmutableSet.of(ssp, sspOtherSystem));

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

    // t = 12: first read for sspOtherSystem, t = 13: first write for sspOtherSystem
    when(clock.currentTimeMillis()).thenReturn(12L, 13L);
    when(otherSystemAdmin.getSSPMetadata(ImmutableSet.of(sspOtherSystem))).thenReturn(
        ImmutableMap.of(sspOtherSystem, sspMetadata(2)));
    assertEquals(sspMetadata(2), cache.getMetadata(sspOtherSystem));
    // does not call for ssp
    verify(otherSystemAdmin).getSSPMetadata(ImmutableSet.of(sspOtherSystem));

    // now entries are stale, do another round of individual fetches
    when(clock.currentTimeMillis()).thenReturn(14 + 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));
    when(otherSystemAdmin.getSSPMetadata(ImmutableSet.of(sspOtherSystem))).thenReturn(
        ImmutableMap.of(sspOtherSystem, sspMetadata(11)));
    assertEquals(sspMetadata(11), cache.getMetadata(sspOtherSystem));
    verify(otherSystemAdmin, times(2)).getSSPMetadata(ImmutableSet.of(sspOtherSystem));
  }
Given that the there are sspsToPrefetch with systems that do not match the requested SSP, getMetadata should not
prefetch all sspsToPrefetch.
samza-core/src/test/java/org/apache/samza/system/TestSSPMetadataCache.java