@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_generalError() {
    // Given:
    ArrayList<PersistentQueryMetadata> queries = new ArrayList<>();
    queries.add(null);
    when(ksqlEngine.getPersistentQueries()).thenReturn(queries);
    SendLagService sendLagService = lagReportingAgent.new SendLagService();

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

    // Then:
    verify(ksqlClient, never()).makeAsyncLagReportRequest(any(), any());
  }