public CompletableFuture<Void> handlePullQuery(
      final ServiceContext serviceContext,
      final PullPhysicalPlan pullPhysicalPlan,
      final ConfiguredStatement<Query> statement,
      final RoutingOptions routingOptions,
      final PullQueryWriteStream pullQueryQueue,
      final CompletableFuture<Void> shouldCancelRequests
  ) {
    final List<KsqlPartitionLocation> allLocations = pullPhysicalPlan.getMaterialization().locator()
        .locate(
            pullPhysicalPlan.getKeys(),
            routingOptions,
            routingFilterFactory,
            pullPhysicalPlan.getPlanType() == PullPhysicalPlanType.RANGE_SCAN
    );

    final Map<Integer, List<Host>> emptyPartitions = allLocations.stream()
        .filter(loc -> loc.getNodes().stream().noneMatch(node -> node.getHost().isSelected()))
        .collect(Collectors.toMap(
            KsqlPartitionLocation::getPartition,
            loc -> loc.getNodes().stream().map(KsqlNode::getHost).collect(Collectors.toList())));

    if (!emptyPartitions.isEmpty()) {
      final MaterializationException materializationException = new MaterializationException(
          "Unable to execute pull query. "
              + emptyPartitions.entrySet()
              .stream()
              .map(kv -> String.format(
                  "Partition %s failed to find valid host. Hosts scanned: %s",
                  kv.getKey(), kv.getValue()))
              .collect(Collectors.joining(", ", "[", "]")));

      LOG.debug(materializationException.getMessage());
      throw materializationException;
    }

    // at this point we should filter out the hosts that we should not route to
    final List<KsqlPartitionLocation> locations = allLocations
        .stream()
        .map(KsqlPartitionLocation::removeFilteredHosts)
        .collect(Collectors.toList());

    final CompletableFuture<Void> completableFuture = new CompletableFuture<>();
    coordinatorExecutorService.submit(() -> {
      try {
        executeRounds(serviceContext, pullPhysicalPlan, statement, routingOptions,
            locations, pullQueryQueue, shouldCancelRequests);
        completableFuture.complete(null);
      } catch (Throwable t) {
        completableFuture.completeExceptionally(t);
      }
    });
    return completableFuture;
  }
@Test
  public void shouldNotRouteToFilteredHost() throws InterruptedException, ExecutionException {
    // Given:
    location1 = new PartitionLocation(Optional.empty(), 1, ImmutableList.of(badNode, node1));
    when(ksqlClient.makeQueryRequest(any(), any(), any(), any(), any(), any(), any()))
        .then(invocationOnMock -> RestResponse.successful(200, 2));
    locate(location1, location2, location3, location4);

    // When:
    final CompletableFuture<Void> fut = haRouting.handlePullQuery(
        serviceContext,
        pullPhysicalPlan,
        statement,
        routingOptions,
        pullQueryQueue,
        disconnect
    );
    fut.get();

    // Then:
    verify(ksqlClient, never())
        .makeQueryRequest(eq(badNode.location()), any(), any(), any(), any(), any(), any());
  }