@VisibleForTesting
  protected KeyQueryMetadata getKeyQueryMetadata(final KsqlKey key) {
    if (sharedRuntimesEnabled && kafkaStreams instanceof KafkaStreamsNamedTopologyWrapper) {
      return ((KafkaStreamsNamedTopologyWrapper) kafkaStreams)
          .queryMetadataForKey(storeName, key.getKey(), keySerializer, queryId);
    }
    try {
      return kafkaStreams.queryMetadataForKey(storeName, key.getKey(), keySerializer);
    } catch (IllegalStateException e) {
      // We may be in `PENDING_SHUTDOWN` here, which means that we should let the client retry,
      // as this will be happening during a roll.
      return KeyQueryMetadata.NOT_AVAILABLE;
    }
  }
@Test
  public void shouldUseNamedTopologyWhenSharedRuntimeIsEnabledForQueryMetadataForKey() {
    // Given:
    final KsLocator locator = new KsLocator(STORE_NAME, kafkaStreamsNamedTopologyWrapper, topology,
        keySerializer, LOCAL_HOST_URL, true, "queryId");

    // When:
    locator.getKeyQueryMetadata(KEY);

    // Then:
    Mockito.verify(kafkaStreamsNamedTopologyWrapper)
        .queryMetadataForKey(STORE_NAME, KEY.getKey(), keySerializer, "queryId");
  }