public StreamPullQueryMetadata createStreamPullQuery(
      final ServiceContext serviceContext,
      final ImmutableAnalysis analysis,
      final ConfiguredStatement<Query> statementOrig,
      final boolean excludeTombstones
  ) {
    final boolean streamPullQueriesEnabled =
        statementOrig
            .getSessionConfig()
            .getConfig(true)
            .getBoolean(KsqlConfig.KSQL_QUERY_STREAM_PULL_QUERY_ENABLED);

    if (!streamPullQueriesEnabled) {
      throw new KsqlStatementException(
          "Pull queries on streams are disabled. To create a push query on the stream,"
              + " add EMIT CHANGES to the end. To enable pull queries on streams, set"
              + " the " + KsqlConfig.KSQL_QUERY_STREAM_PULL_QUERY_ENABLED + " config to 'true'.",
          statementOrig.getMaskedStatementText()
      );
    }

    // Stream pull query overrides.
    final Map<String, Object> overrides =
        new HashMap<>(statementOrig.getSessionConfig().getOverrides());
    // Starting from earliest is semantically necessary.
    overrides.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    // Using a single thread keeps these queries as lightweight as possible, since we are
    // not counting them against the transient query limit.
    overrides.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
    // There's no point in EOS, since this query only produces side effects.
    overrides.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_LEAST_ONCE);

    final ConfiguredStatement<Query> statement = statementOrig.withConfigOverrides(overrides);
    final ImmutableMap<TopicPartition, Long> endOffsets =
        getQueryInputEndOffsets(analysis, serviceContext.getAdminClient());

    final TransientQueryMetadata transientQueryMetadata = EngineExecutor
        .create(primaryContext, serviceContext, statement.getSessionConfig())
        .executeStreamPullQuery(statement, excludeTombstones, endOffsets);

    QueryLogger.info(
        "Streaming stream pull query results '{}' from earliest to " + endOffsets,
        statement.getMaskedStatementText()
    );

    return new StreamPullQueryMetadata(transientQueryMetadata, endOffsets);
  }
@Test
  public void shouldCheckStreamPullQueryEnabledFlag() {
    setupKsqlEngineWithSharedRuntimeEnabled();
    @SuppressWarnings("unchecked") final ConfiguredStatement<Query> statementOrig =
        mock(ConfiguredStatement.class);
    when(statementOrig.getMaskedStatementText()).thenReturn("TEXT");

    final SessionConfig mockSessionConfig = mock(SessionConfig.class);
    final KsqlConfig mockConfig = mock(KsqlConfig.class);
    when(statementOrig.getSessionConfig()).thenReturn(mockSessionConfig);
    when(mockSessionConfig.getConfig(eq(true))).thenReturn(mockConfig);
    when(mockConfig.getBoolean(eq(KsqlConfig.KSQL_QUERY_STREAM_PULL_QUERY_ENABLED)))
        .thenReturn(false);

    final KsqlStatementException ksqlStatementException = assertThrows(
        KsqlStatementException.class,
        () -> ksqlEngine.createStreamPullQuery(
            null,
            null,
            statementOrig,
            false
        )
    );

    assertThat(ksqlStatementException.getSqlStatement(), is("TEXT"));
    assertThat(ksqlStatementException.getRawMessage(), is(
        "Pull queries on streams are disabled."
            + " To create a push query on the stream, add EMIT CHANGES to the end."
            + " To enable pull queries on streams,"
            + " set the ksql.query.pull.stream.enabled config to 'true'."
    ));
  }