@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_ignores_NodeHealth_of_more_than_30_seconds_before_cluster_time() {
    NodeHealth nodeHealth = randomNodeHealth();
    Map<UUID, TimestampedNodeHealth> map = new HashMap<>();
    UUID memberUuid = UUID.randomUUID();
    TimestampedNodeHealth timestampedNodeHealth = new TimestampedNodeHealth(nodeHealth, clusterTime - 30 * 1000 - random.nextInt(99));
    map.put(memberUuid, timestampedNodeHealth);
    doReturn(map).when(hazelcastMember).getReplicatedMap(MAP_SQ_HEALTH_STATE);
    when(hazelcastMember.getMemberUuids()).thenReturn(map.keySet());
    when(hazelcastMember.getClusterTime()).thenReturn(clusterTime);

    assertThat(underTest.readAll()).isEmpty();
  }