public List<KsqlNode> locate() {
    final List<PersistentQueryMetadata> currentQueries = allPersistentQueries.get();
    if (currentQueries.isEmpty()) {
      return Collections.emptyList();
    }

    return currentQueries.stream()
        .map(QueryMetadata::getAllStreamsHostMetadata)
        .filter(Objects::nonNull)
        .flatMap(Collection::stream)
        .map(StreamsMetadata::hostInfo)
        .map(hi -> new Node(isLocalhost(hi), buildLocation(hi)))
        .distinct()
        .collect(Collectors.toList());
  }
@Test
  public void shouldLocate() throws MalformedURLException {
    // Given:
    final AllHostsLocator locator = new AllHostsLocator(
        () -> ImmutableList.of(metadata1, metadata2),
        new URL("http://localhost:8088")
    );
    when(metadata1.getAllStreamsHostMetadata())
        .thenReturn(ImmutableList.of(streamsMetadata1, streamsMetadata2));
    when(metadata2.getAllStreamsHostMetadata())
        .thenReturn(ImmutableList.of(streamsMetadata3));
    when(streamsMetadata1.hostInfo())
        .thenReturn(new HostInfo("abc", 101), new HostInfo("localhost", 8088));
    when(streamsMetadata2.hostInfo()).thenReturn(new HostInfo("localhost", 8088));
    when(streamsMetadata3.hostInfo()).thenReturn(new HostInfo("localhost", 8089));

    // When:
    final List<KsqlNode> nodes = ImmutableList.copyOf(locator.locate());

    // Then:
    assertThat(nodes.size(), is(3));
    assertThat(nodes.get(0).isLocal(), is(false));
    assertThat(nodes.get(0).location().toString(), is("http://abc:101/"));
    assertThat(nodes.get(1).isLocal(), is(true));
    assertThat(nodes.get(1).location().toString(), is("http://localhost:8088/"));
    assertThat(nodes.get(2).isLocal(), is(false));
    assertThat(nodes.get(2).location().toString(), is("http://localhost:8089/"));
  }