@SuppressWarnings("ParameterNumber")
  TransientQueryMetadata buildTransientQuery(
      final String statementText,
      final QueryId queryId,
      final Set<SourceName> sources,
      final ExecutionStep<?> physicalPlan,
      final String planSummary,
      final LogicalSchema schema,
      final OptionalInt limit,
      final Optional<WindowInfo> windowInfo,
      final boolean excludeTombstones,
      final QueryMetadata.Listener listener,
      final StreamsBuilder streamsBuilder,
      final Optional<ImmutableMap<TopicPartition, Long>> endOffsets,
      final MetricCollectors metricCollectors
  ) {
    final KsqlConfig ksqlConfig = config.getConfig(true);
    final String applicationId = QueryApplicationId.build(ksqlConfig, false, queryId);
    final RuntimeBuildContext runtimeBuildContext = buildContext(
        applicationId,
        queryId,
        streamsBuilder
    );

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

    streamsProperties.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 0);

    final Object buildResult = buildQueryImplementation(physicalPlan, runtimeBuildContext);
    final TransientQueryQueue queue =
        buildTransientQueryQueue(buildResult, limit, excludeTombstones, endOffsets);
    final Topology topology = streamsBuilder.build(PropertiesUtil.asProperties(streamsProperties));

    final TransientQueryMetadata.ResultType resultType = buildResult instanceof KTableHolder
        ? windowInfo.isPresent() ? ResultType.WINDOWED_TABLE : ResultType.TABLE
        : ResultType.STREAM;

    return new TransientQueryMetadata(
        statementText,
        schema,
        sources,
        planSummary,
        queue,
        queryId,
        applicationId,
        topology,
        kafkaStreamsBuilder,
        streamsProperties,
        config.getOverrides(),
        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG),
        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
        resultType,
        ksqlConfig.getLong(KsqlConfig.KSQL_QUERY_RETRY_BACKOFF_INITIAL_MS),
        ksqlConfig.getLong(KsqlConfig.KSQL_QUERY_RETRY_BACKOFF_MAX_MS),
        listener,
        processingLogContext.getLoggerFactory()
    );
  }
@Test
  public void shouldBuildTransientQueryWithSharedRutimesCorrectly() {
    // Given:
    givenTransientQuery();
    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_SHARED_RUNTIME_ENABLED)).thenReturn(true);

    // When:
    final TransientQueryMetadata queryMetadata = queryBuilder.buildTransientQuery(
        STATEMENT_TEXT,
        QUERY_ID,
        SOURCES.stream().map(DataSource::getName).collect(Collectors.toSet()),
        physicalPlan,
        SUMMARY,
        TRANSIENT_SINK_SCHEMA,
        LIMIT,
        Optional.empty(),
        false,
        queryListener,
        streamsBuilder,
        Optional.empty(),
        new MetricCollectors()
    );
    queryMetadata.initialize();

    // Then:
    assertThat(queryMetadata.getStatementString(), equalTo(STATEMENT_TEXT));
    assertThat(queryMetadata.getSourceNames(), equalTo(SOURCES.stream()
        .map(DataSource::getName).collect(Collectors.toSet())));
    assertThat(queryMetadata.getExecutionPlan(), equalTo(SUMMARY));
    assertThat(queryMetadata.getTopology(), is(topology));
    assertThat(queryMetadata.getOverriddenProperties(), equalTo(OVERRIDES));
    verify(kafkaStreamsBuilder).build(any(), propertyCaptor.capture());
    assertThat(queryMetadata.getStreamsProperties(), equalTo(propertyCaptor.getValue()));
    assertThat(queryMetadata.getStreamsProperties().get(InternalConfig.TOPIC_PREFIX_ALTERNATIVE), nullValue());
  }