public Pair<Map<HostInfo, KsqlEntity>, Set<HostInfo>> fetchAllRemoteResults() {
    final Set<HostInfo> remoteHosts = DiscoverRemoteHostsUtil.getRemoteHosts(
        executionContext.getPersistentQueries(),
        sessionProperties.getKsqlHostInfo()
    );
    if (remoteHosts.isEmpty() || sessionProperties.getInternalRequest()) {
      return new Pair<>(ImmutableMap.of(), ImmutableSet.of());
    }

    final Set<HostInfo> unresponsiveHosts = new HashSet<>();
    final ExecutorService executorService = Executors.newFixedThreadPool(remoteHosts.size());

    try {
      final Map<HostInfo, CompletableFuture<RestResponse<KsqlEntityList>>> futureResponses =
          new HashMap<>();
      for (HostInfo host : remoteHosts) {
        futureResponses.put(host, fetchRemoteData(statement.getUnMaskedStatementText(), host,
            executorService));
      }

      final ImmutableMap.Builder<HostInfo, KsqlEntity> results = ImmutableMap.builder();
      for (final Map.Entry<HostInfo, CompletableFuture<RestResponse<KsqlEntityList>>> e
          : futureResponses.entrySet()) {
        try {
          final RestResponse<KsqlEntityList> response =
              e.getValue().get(
                  executionContext.getKsqlConfig().getLong(KSQL_FETCH_REMOTE_HOSTS_TIMEOUT_SECONDS),
                  TimeUnit.SECONDS);
          if (response.isErroneous()) {
            LOG.warn("Error response from host. host: {}, cause: {}",
                e.getKey(), response.getErrorMessage().getMessage());
            unresponsiveHosts.add(e.getKey());
          } else {
            results.put(e.getKey(), response.getResponse().get(0));
          }
        } catch (final Exception cause) {
          LOG.warn("Failed to retrieve info from host: {}, statement: {}, cause: {}",
              e.getKey(), statement.getMaskedStatementText(), cause);
          unresponsiveHosts.add(e.getKey());
        }
      }

      return new Pair<>(results.build(), unresponsiveHosts);
    } finally {
      executorService.shutdown();
    }
  }
@Test
  public void testReturnsEmptyIfRequestIsInternal() {
    Pair<Map<HostInfo, KsqlEntity>, Set<HostInfo>> remoteResults = augmenter.fetchAllRemoteResults();
    assertThat(remoteResults.getLeft().entrySet(), hasSize(0));
    assertThat(remoteResults.getRight(), hasSize(0));
  }