@SuppressWarnings("ParameterNumber")
  PersistentQueryMetadata buildPersistentQueryInDedicatedRuntime(
      final KsqlConfig ksqlConfig,
      final KsqlConstants.PersistentQueryType persistentQueryType,
      final String statementText,
      final QueryId queryId,
      final Optional<DataSource> sinkDataSource,
      final Set<DataSource> sources,
      final ExecutionStep<?> physicalPlan,
      final String planSummary,
      final QueryMetadata.Listener listener,
      final Supplier<List<PersistentQueryMetadata>> allPersistentQueries,
      final StreamsBuilder streamsBuilder,
      final MetricCollectors metricCollectors) {

    final String applicationId = QueryApplicationId.build(ksqlConfig, true, queryId);
    final Map<String, Object> streamsProperties = buildStreamsProperties(
        applicationId,
        Optional.of(queryId),
        metricCollectors,
        config.getConfig(true),
        processingLogContext
    );

    final LogicalSchema logicalSchema;
    final KeyFormat keyFormat;
    final ValueFormat valueFormat;
    final KsqlTopic ksqlTopic;

    switch (persistentQueryType) {
      // CREATE_SOURCE does not have a sink, so the schema is obtained from the query source
      case CREATE_SOURCE:
        final DataSource dataSource = Iterables.getOnlyElement(sources);

        logicalSchema = dataSource.getSchema();
        keyFormat = dataSource.getKsqlTopic().getKeyFormat();
        valueFormat = dataSource.getKsqlTopic().getValueFormat();
        ksqlTopic = dataSource.getKsqlTopic();

        break;
      default:
        logicalSchema = sinkDataSource.get().getSchema();
        keyFormat = sinkDataSource.get().getKsqlTopic().getKeyFormat();
        valueFormat = sinkDataSource.get().getKsqlTopic().getValueFormat();
        ksqlTopic = sinkDataSource.get().getKsqlTopic();

        break;
    }

    final PhysicalSchema querySchema = PhysicalSchema.from(
        logicalSchema,
        keyFormat.getFeatures(),
        valueFormat.getFeatures()
    );

    final RuntimeBuildContext runtimeBuildContext = buildContext(
        applicationId,
        queryId,
        streamsBuilder
    );
    final Object result = buildQueryImplementation(physicalPlan, runtimeBuildContext);
    final Topology topology = streamsBuilder
            .build(PropertiesUtil.asProperties(streamsProperties));

    final Optional<MaterializationProviderBuilderFactory.MaterializationProviderBuilder>
        materializationProviderBuilder = getMaterializationInfo(result).map(info ->
            materializationProviderBuilderFactory.materializationProviderBuilder(
                info,
                querySchema,
                keyFormat,
                streamsProperties,
                applicationId,
                queryId.toString()
            ));

    final Optional<ScalablePushRegistry> scalablePushRegistry = applyScalablePushProcessor(
        querySchema.logicalSchema(),
        result,
        allPersistentQueries,
        streamsProperties,
        applicationId,
        ksqlConfig,
        ksqlTopic,
        serviceContext
    );

    return new PersistentQueryMetadataImpl(
        persistentQueryType,
        statementText,
        querySchema,
        sources.stream().map(DataSource::getName).collect(Collectors.toSet()),
        sinkDataSource,
        planSummary,
        queryId,
        materializationProviderBuilder,
        applicationId,
        topology,
        kafkaStreamsBuilder,
        runtimeBuildContext.getSchemas(),
        streamsProperties,
        config.getOverrides(),
        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG),
        getConfiguredQueryErrorClassifier(ksqlConfig, applicationId),
        physicalPlan,
        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
        getUncaughtExceptionProcessingLogger(queryId),
        ksqlConfig.getLong(KsqlConfig.KSQL_QUERY_RETRY_BACKOFF_INITIAL_MS),
        ksqlConfig.getLong(KsqlConfig.KSQL_QUERY_RETRY_BACKOFF_MAX_MS),
        listener,
        scalablePushRegistry,
        processingLogContext.getLoggerFactory()
    );

  }
@Test
  public void shouldBuildDedicatedCreateAsPersistentQueryWithSharedRuntimeCorrectly() {
    // Given:
    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_SHARED_RUNTIME_ENABLED)).thenReturn(true);
    final ProcessingLogger uncaughtProcessingLogger = mock(ProcessingLogger.class);
    when(processingLoggerFactory.getLogger(
        QueryLoggerUtil.queryLoggerName(QUERY_ID, new QueryContext.Stacker()
            .push("ksql.logger.thread.exception.uncaught").getQueryContext()),
        Collections.singletonMap("query-id", QUERY_ID.toString()))
    ).thenReturn(uncaughtProcessingLogger);

    // When:
    final PersistentQueryMetadata queryMetadata =  queryBuilder.buildPersistentQueryInDedicatedRuntime(
        ksqlConfig,
        KsqlConstants.PersistentQueryType.CREATE_AS,
        STATEMENT_TEXT,
        QUERY_ID,
        Optional.of(sink),
        SOURCES,
        physicalPlan,
        SUMMARY,
        queryListener,
        ArrayList::new,
        streamsBuilder,
        new MetricCollectors()
    );
    queryMetadata.initialize();

    // Then:
    assertThat(queryMetadata.getStatementString(), equalTo(STATEMENT_TEXT));
    assertThat(queryMetadata.getQueryId(), equalTo(QUERY_ID));
    assertThat(queryMetadata.getSinkName().get(), equalTo(SINK_NAME));
    assertThat(queryMetadata.getPhysicalSchema(), equalTo(SINK_PHYSICAL_SCHEMA));
    assertThat(queryMetadata.getResultTopic(), is(Optional.of(ksqlTopic)));
    assertThat(queryMetadata.getSourceNames(), equalTo(SOURCES.stream()
        .map(DataSource::getName).collect(Collectors.toSet())));
    assertThat(queryMetadata.getDataSourceType().get(), equalTo(DataSourceType.KSTREAM));
    assertThat(queryMetadata.getExecutionPlan(), equalTo(SUMMARY));
    assertThat(queryMetadata.getTopology(), is(topology));
    assertThat(queryMetadata.getOverriddenProperties(), equalTo(OVERRIDES));
    assertThat(queryMetadata.getProcessingLogger(), equalTo(uncaughtProcessingLogger));
    assertThat(queryMetadata.getPersistentQueryType(),
        equalTo(KsqlConstants.PersistentQueryType.CREATE_AS));
    // queries in dedicated runtimes must not include alternative topic prefix
    assertThat(
        queryMetadata.getStreamsProperties().get(InternalConfig.TOPIC_PREFIX_ALTERNATIVE),
        is(nullValue())
    );
  }