public Optional<KsMaterialization> create(
      final String stateStoreName,
      final KafkaStreams kafkaStreams,
      final Topology topology,
      final LogicalSchema schema,
      final Serializer<GenericKey> keySerializer,
      final Optional<WindowInfo> windowInfo,
      final Map<String, ?> streamsProperties,
      final KsqlConfig ksqlConfig,
      final String applicationId,
      final String queryId
  ) {
    final Object appServer = streamsProperties.get(StreamsConfig.APPLICATION_SERVER_CONFIG);
    if (appServer == null) {
      return Optional.empty();
    }

    final URL localHost = buildLocalHost(appServer);

    final KsLocator locator = locatorFactory.create(
        stateStoreName,
        kafkaStreams,
        topology,
        keySerializer,
        localHost,
        ksqlConfig.getBoolean(KsqlConfig.KSQL_SHARED_RUNTIME_ENABLED),
        queryId
    );

    final KsStateStore stateStore = storeFactory.create(
        stateStoreName,
        kafkaStreams,
        schema,
        ksqlConfig,
        queryId
    );

    final KsMaterialization materialization = materializationFactory.create(
        windowInfo,
        locator,
        stateStore
    );

    return Optional.of(materialization);
  }
@Test
  public void shouldBuildLocatorWithCorrectParams() {
    // When:
    factory.create(STORE_NAME, kafkaStreams, topology, SCHEMA, keySerializer, Optional.empty(),
        streamsProperties, ksqlConfig, APPLICATION_ID, "queryId");

    // Then:
    verify(locatorFactory).create(
        STORE_NAME,
        kafkaStreams,
        topology,
        keySerializer,
        DEFAULT_APP_SERVER,
        false,
        "queryId"
    );
  }