@Override
  public List<KsqlPartitionLocation> locate(
      final List<KsqlKey> keys,
      final RoutingOptions routingOptions,
      final RoutingFilterFactory routingFilterFactory,
      final boolean isRangeScan
  ) {
    if (isRangeScan && keys.isEmpty()) {
      throw new IllegalStateException("Query is range scan but found no range keys.");
    }
    final ImmutableList.Builder<KsqlPartitionLocation> partitionLocations = ImmutableList.builder();
    final Set<Integer> filterPartitions = routingOptions.getPartitions();
    final Optional<Set<KsqlKey>> keySet = keys.isEmpty() ? Optional.empty() :
        Optional.of(Sets.newHashSet(keys));

    // Depending on whether this is a key-based lookup, determine which metadata method to use.
    // If we don't have keys, find the metadata for all partitions since we'll run the query for
    // all partitions of the state store rather than a particular one.
    //For issue #7174. Temporarily turn off metadata finding for a partition with keys
    //if there are more than one key.
    final List<PartitionMetadata> metadata;
    if (keys.size() == 1 && keys.get(0).getKey().size() == 1 && !isRangeScan) {
      metadata = getMetadataForKeys(keys, filterPartitions);
    } else {
      metadata = getMetadataForAllPartitions(filterPartitions, keySet);
    }

    if (metadata.isEmpty()) {
      final MaterializationException materializationException = new MaterializationException(
          "Cannot determine which host contains the required partitions to serve the pull query. \n"
              + "The underlying persistent query may be restarting (e.g. as a result of "
              + "ALTER SYSTEM) view the status of your by issuing <DESCRIBE foo>.");
      LOG.debug(materializationException.getMessage());
      throw materializationException;
    }

    // Go through the metadata and group them by partition.
    for (PartitionMetadata partitionMetadata : metadata) {
      LOG.debug("Handling pull query for partition {} of state store {}.",
                partitionMetadata.getPartition(), storeName);
      final HostInfo activeHost = partitionMetadata.getActiveHost();
      final Set<HostInfo> standByHosts = partitionMetadata.getStandbyHosts();
      final int partition = partitionMetadata.getPartition();
      final Optional<Set<KsqlKey>> partitionKeys = partitionMetadata.getKeys();
      LOG.debug("Active host {}, standby {}, partition {}.",
               activeHost, standByHosts, partition);
      // For a given partition, find the ordered, filtered list of hosts to consider
      final List<KsqlNode> filteredHosts = getFilteredHosts(routingOptions, routingFilterFactory,
          activeHost, standByHosts, partition);

      partitionLocations.add(new PartitionLocation(partitionKeys, partition, filteredHosts));
    }
    return partitionLocations.build();
  }
@Test
  public void shouldFindAllPartitionsWhenNoKeys() {
    // Given:
    when(topology.describe()).thenReturn(description);
    when(description.subtopologies()).thenReturn(ImmutableSet.of(sub1));
    when(sub1.nodes()).thenReturn(ImmutableSet.of(source, processor));
    when(source.topicSet()).thenReturn(ImmutableSet.of(TOPIC_NAME));
    when(processor.stores()).thenReturn(ImmutableSet.of(STORE_NAME));
    when(kafkaStreams.streamsMetadataForStore(any()))
        .thenReturn(ImmutableList.of(HOST1_STREAMS_MD1, HOST1_STREAMS_MD2, HOST1_STREAMS_MD3));

    // When:
    final List<KsqlPartitionLocation> result = locator.locate(
        ImmutableList.of(), routingOptions, routingFilterFactoryStandby, false);

    // Then:
    assertThat(result.size(), is(3));
    int partition = result.get(0).getPartition();
    assertThat(partition, is(0));
    List<KsqlNode> nodeList = result.get(0).getNodes();
    assertThat(nodeList.size(), is(3));
    assertThat(nodeList.get(0), is(activeNode));
    assertThat(nodeList.get(1), is(standByNode1));
    assertThat(nodeList.get(2), is(standByNode2));
    partition = result.get(1).getPartition();
    assertThat(partition, is(1));
    nodeList = result.get(1).getNodes();
    assertThat(nodeList.size(), is(3));
    assertThat(nodeList.get(0), is(standByNode1));
    assertThat(nodeList.get(1), is(activeNode));
    assertThat(nodeList.get(2), is(standByNode2));
    partition = result.get(2).getPartition();
    assertThat(partition, is(2));
    nodeList = result.get(2).getNodes();
    assertThat(nodeList.size(), is(3));
    assertThat(nodeList.get(0), is(standByNode2));
    assertThat(nodeList.get(1), is(activeNode));
    assertThat(nodeList.get(2), is(standByNode1));
  }