@Test
  public void testGetMetadataConcurrentAccess() throws ExecutionException, InterruptedException {
    int numPartitions = 50;
    // initial fetch
    when(clock.currentTimeMillis()).thenReturn(10L);
    Set<SystemStreamPartition> ssps =
        IntStream.range(0, numPartitions).mapToObj(TestSSPMetadataCache::buildSSP).collect(Collectors.toSet());
    SSPMetadataCache cache = buildSSPMetadataCache(ssps);
    ExecutorService executorService = Executors.newFixedThreadPool(10);
    when(systemAdmin.getSSPMetadata(ssps)).thenAnswer(invocation -> {
      // have the admin call wait so that it forces the threads to overlap on the lock
      Thread.sleep(500);
      return IntStream.range(0, numPartitions)
          .boxed()
          .collect(Collectors.toMap(TestSSPMetadataCache::buildSSP, i -> sspMetadata((long) i)));
    });

    // send concurrent requests for metadata
    List<Future<SystemStreamMetadata.SystemStreamPartitionMetadata>> getMetadataFutures =
        IntStream.range(0, numPartitions)
            .mapToObj(i -> executorService.submit(() -> cache.getMetadata(buildSSP(i))))
            .collect(Collectors.toList());
    for (int i = 0; i < numPartitions; i++) {
      assertEquals(sspMetadata(i), getMetadataFutures.get(i).get());
    }
    // should only see one call to fetch metadata
    verify(systemAdmin).getSSPMetadata(ssps);

    // make entries stale
    when(clock.currentTimeMillis()).thenReturn(11 + CACHE_TTL.toMillis());
    getMetadataFutures = IntStream.range(0, numPartitions)
        .mapToObj(i -> executorService.submit(() -> cache.getMetadata(buildSSP(i))))
        .collect(Collectors.toList());
    for (int i = 0; i < numPartitions; i++) {
      assertEquals(sspMetadata(i), getMetadataFutures.get(i).get());
    }
    // should see two total calls to fetch metadata
    verify(systemAdmin, times(2)).getSSPMetadata(ssps);
  }
Given concurrent access to getMetadata, there should be only single calls to fetch metadata.
samza-core/src/test/java/org/apache/samza/system/TestSSPMetadataCache.java