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 shouldCallRouteQuery_allStandbysFail() {
    // Given:
    locate(location1, location2, location3, location4);
    doAnswer(i -> {
      throw new StandbyFallbackException("Error1!");
    }).when(pullPhysicalPlan).execute(eq(ImmutableList.of(location1, location3)), any(), any());
    when(ksqlClient.makeQueryRequest(eq(node2.location()), any(), any(), any(), any(), any(), any()))
        .thenAnswer(new Answer() {
          private int count = 0;

          public Object answer(InvocationOnMock i) {
            Map<String, ?> requestProperties = i.getArgument(3);
            PullQueryWriteStream rowConsumer = i.getArgument(4);
            if (++count == 1) {
              assertThat(requestProperties.get(
                  KsqlRequestConfig.KSQL_REQUEST_QUERY_PULL_PARTITIONS), is("2,4"));
              rowConsumer.write(
                  ImmutableList.of(
                      StreamedRow.header(queryId, logicalSchema),
                      StreamedRow.pullRow(GenericRow.fromList(ROW2), Optional.empty())));
            } else {
              assertThat(requestProperties.get(
                  KsqlRequestConfig.KSQL_REQUEST_QUERY_PULL_PARTITIONS), is("1,3"));
              throw new RuntimeException("Error2!");
            }

            return RestResponse.successful(200, 2);
          }
        }
    );
    // When:
    final Exception e = assertThrows(
        ExecutionException.class,
        () -> {
          CompletableFuture<Void> future = haRouting.handlePullQuery(serviceContext,
              pullPhysicalPlan, statement, routingOptions, pullQueryQueue,
              disconnect);
          future.get();
        }
    );

    // Then:
    verify(pullPhysicalPlan).execute(eq(ImmutableList.of(location1, location3)), any(), any());
    verify(ksqlClient, times(2)).makeQueryRequest(eq(node2.location()), any(), any(), any(), any(), any(), any());
    assertThat(e.getCause().getMessage(), containsString("Exhausted standby hosts to try."));
  }