class QueryExecutor {

  public 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 BlockingRowQueue queue = buildTransientQueryQueue(queryId, physicalPlan, limit);

    final KsqlConfig ksqlConfig = config.getConfig(true);

    final String applicationId = QueryApplicationId.build(ksqlConfig, false, queryId);

    final Map<String, Object> streamsProperties = buildStreamsProperties(applicationId, queryId);
    final Topology topology = streamsBuilder.build(PropertiesUtil.asProperties(streamsProperties));

    return new TransientQueryMetadata(
        statementText,
        schema,
        sources,
        planSummary,
        queue,
        applicationId,
        topology,
        kafkaStreamsBuilder,
        streamsProperties,
        config.getOverrides(),
        queryCloseCallback,
        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG),
        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE)
    );
  }

  public  QueryExecutor(
      final SessionConfig config,
      final ProcessingLogContext processingLogContext,
      final ServiceContext serviceContext,
      final FunctionRegistry functionRegistry,
      final Consumer<QueryMetadata> queryCloseCallback);
  @VisibleForTesting  QueryExecutor(
      final SessionConfig config,
      final ProcessingLogContext processingLogContext,
      final ServiceContext serviceContext,
      final FunctionRegistry functionRegistry,
      final Consumer<QueryMetadata> queryCloseCallback,
      final KafkaStreamsBuilder kafkaStreamsBuilder,
      final StreamsBuilder streamsBuilder,
      final MaterializationProviderBuilderFactory materializationProviderBuilderFactory
  );

}