public static Set<HostInfo> getRemoteHosts(
      final List<PersistentQueryMetadata> currentQueries, 
      final KsqlHostInfo localHost
  ) {
    return currentQueries.stream()
        // required filter else QueryMetadata.getAllMetadata() throws
        .filter(q -> q.getState().isRunningOrRebalancing())
        .map(QueryMetadata::getAllStreamsHostMetadata)
        .filter(Objects::nonNull)
        .flatMap(Collection::stream)
        .map(StreamsMetadata::hostInfo)
        .filter(hostInfo -> !(hostInfo.host().equals(localHost.host())
            && hostInfo.port() == (localHost.port())))
        .collect(Collectors.toSet());
  }
@Test
  public void shouldFilterQueryMetadataByState() {
    // When:
    final Set<HostInfo> info = DiscoverRemoteHostsUtil.getRemoteHosts(
        ImmutableList.of(runningQuery, notRunningQuery),
        THIS_HOST_INFO
    );

    // Then:
    assertThat(info, contains(OTHER_HOST_INFO));

    verify(notRunningQuery, never()).getAllStreamsHostMetadata();
  }