class QueryExecutor {

  public PersistentQueryMetadata buildPersistentQuery(
      final String statementText,
      final QueryId queryId,
      final DataSource sinkDataSource,
      final Set<SourceName> sources,
      final ExecutionStep<?> physicalPlan,
      final String planSummary
  ) {
    final KsqlQueryBuilder ksqlQueryBuilder = queryBuilder(queryId);
    final PlanBuilder planBuilder = new KSPlanBuilder(ksqlQueryBuilder);
    final Object result = physicalPlan.build(planBuilder);

    final KsqlConfig ksqlConfig = config.getConfig(true);

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

    final PhysicalSchema querySchema = PhysicalSchema.from(
        sinkDataSource.getSchema(),
        sinkDataSource.getKsqlTopic().getKeyFormat().getFeatures(),
        sinkDataSource.getKsqlTopic().getValueFormat().getFeatures()
    );

    final Optional<MaterializationProviderBuilderFactory.MaterializationProviderBuilder>
        materializationProviderBuilder = getMaterializationInfo(result).map(info ->
            materializationProviderBuilderFactory.materializationProviderBuilder(
                info,
                querySchema,
                sinkDataSource.getKsqlTopic().getKeyFormat(),
                streamsProperties,
                applicationId
            ));

    final QueryErrorClassifier topicClassifier = new MissingTopicClassifier(applicationId);
    final QueryErrorClassifier classifier = buildConfiguredClassifiers(ksqlConfig, applicationId)
        .map(topicClassifier::and)
        .orElse(topicClassifier);

    return new PersistentQueryMetadata(
        statementText,
        querySchema,
        sources,
        sinkDataSource,
        planSummary,
        queryId,
        materializationProviderBuilder,
        applicationId,
        topology,
        kafkaStreamsBuilder,
        ksqlQueryBuilder.getSchemas(),
        streamsProperties,
        config.getOverrides(),
        queryCloseCallback,
        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG),
        classifier,
        physicalPlan,
        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
        getUncaughtExceptionProcessingLogger(queryId)
    );
  }

  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
  );

}