@Override
  public Set<NodeHealth> readAll() {
    long clusterTime = hzMember.getClusterTime();
    long timeout = clusterTime - TIMEOUT_30_SECONDS;
    Map<UUID, TimestampedNodeHealth> sqHealthState = readReplicatedMap();
    Set<UUID> hzMemberUUIDs = hzMember.getMemberUuids();
    Set<NodeHealth> existingNodeHealths = sqHealthState.entrySet().stream()
      .filter(outOfDate(timeout))
      .filter(ofNonExistentMember(hzMemberUUIDs))
      .map(entry -> entry.getValue().getNodeHealth())
      .collect(Collectors.toSet());
    if (LOG.isTraceEnabled()) {
      LOG.trace("Reading {} and keeping {}", new HashMap<>(sqHealthState), existingNodeHealths);
    }
    return ImmutableSet.copyOf(existingNodeHealths);
  }
@Test
  public void readAll_returns_all_NodeHealth_in_map_sq_health_state_for_existing_client_uuids_aged_less_than_30_seconds() {
    NodeHealth[] nodeHealths = IntStream.range(0, 1 + random.nextInt(6)).mapToObj(i -> randomNodeHealth()).toArray(NodeHealth[]::new);
    Map<UUID, TimestampedNodeHealth> allNodeHealths = new HashMap<>();
    Map<UUID, NodeHealth> expected = new HashMap<>();
    for (NodeHealth nodeHealth : nodeHealths) {
      UUID memberUuid = UUID.randomUUID();
      TimestampedNodeHealth timestampedNodeHealth = new TimestampedNodeHealth(nodeHealth, clusterTime - random.nextInt(30 * 1000));
      allNodeHealths.put(memberUuid, timestampedNodeHealth);
      if (random.nextBoolean()) {
        expected.put(memberUuid, nodeHealth);
      }
    }
    doReturn(allNodeHealths).when(hazelcastMember).getReplicatedMap(MAP_SQ_HEALTH_STATE);
    when(hazelcastMember.getMemberUuids()).thenReturn(expected.keySet());
    when(hazelcastMember.getClusterTime()).thenReturn(clusterTime);

    assertThat(underTest.readAll())
      .containsOnly(expected.values().toArray(new NodeHealth[0]));
    assertThat(logging.getLogs()).isEmpty();
  }