@Override
  public void onHostStatusUpdated(final Map<KsqlHostInfo, HostStatus> hostsStatusMap) {
    aliveHostsRef.set(hostsStatusMap.entrySet().stream()
        .filter(entry -> entry.getValue().isHostAlive())
        .map(Entry::getKey)
        .collect(ImmutableSet.toImmutableSet()));
  }
@Test
  public void shouldSendLags() {
    // Given:
    setupLags();
    SendLagService sendLagService = lagReportingAgent.new SendLagService();

    // When:
    lagReportingAgent.onHostStatusUpdated(HOSTS_ALIVE);
    sendLagService.runOneIteration();

    // Then:
    LagReportingMessage exp = hostLag(LOCALHOST_INFO, LAG_MAP2, TIME_NOW_MS);
    verify(ksqlClient).makeAsyncLagReportRequest(eq(URI.create("http://host2:1234/")), eq(exp));
    verify(ksqlClient).makeAsyncLagReportRequest(eq(URI.create("http://host1:1234/")), eq(exp));
  }