@Override
  public Host filter(final KsqlHostInfo host) {

    if (!heartbeatAgent.isPresent()) {
      return Host.include(host);
    }

    final Map<KsqlHostInfo, HostStatus> allHostsStatus = heartbeatAgent.get().getHostsStatus();
    final HostStatus status = allHostsStatus.get(host);
    if (status == null) {
      return Host.include(host);
    }

    if (status.isHostAlive()) {
      return Host.include(host);
    } else {
      return Host.exclude(host, "Host is not alive as of time " + status.getLastStatusUpdateMs());
    }
  }
@Test
  public void shouldFilterAllDead() {
    // Given:
    allHostsStatus = ImmutableMap.of(
        activeHost, HOST_DEAD,
        standByHost1, HOST_DEAD,
        standByHost2, HOST_DEAD
    );
    when(heartbeatAgent.getHostsStatus()).thenReturn(allHostsStatus);

    // When:
    final Host filterActive = livenessFilter.filter(activeHost);
    final Host filterStandby1 = livenessFilter.filter(standByHost1);
    final Host filterStandby2 = livenessFilter.filter(standByHost2);

    // Then:
    assertThat(filterActive.isSelected(), is(false));
    assertThat(filterActive.getReasonNotSelected(), is("Host is not alive as of time 0"));

    assertThat(filterStandby1.isSelected(), is(false));
    assertThat(filterActive.getReasonNotSelected(), is("Host is not alive as of time 0"));

    assertThat(filterStandby2.isSelected(), is(false));
    assertThat(filterActive.getReasonNotSelected(), is("Host is not alive as of time 0"));
  }