void handleStatement(final QueuedCommand queuedCommand) {
    throwIfNotConfigured();

    handleStatementWithTerminatedQueries(
        queuedCommand.getAndDeserializeCommand(commandDeserializer),
        queuedCommand.getAndDeserializeCommandId(),
        queuedCommand.getStatus(),
        Mode.EXECUTE,
        queuedCommand.getOffset(),
        false
    );
  }
@Test
  public void shouldBuildQueriesWithPersistedConfig() {
    // Given:
    final KsqlConfig originalConfig = new KsqlConfig(
        Collections.singletonMap(
            KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG, "not-the-default"));

    // get a statement instance
    final String ddlText
        = "CREATE STREAM pageviews (viewtime bigint, pageid varchar) " +
        "WITH (kafka_topic='pageview_topic', KEY_FORMAT='kafka', VALUE_FORMAT='json');";
    final String statementText
        = "CREATE STREAM user1pv AS select * from pageviews WHERE userid = 'user1';";
    final PreparedStatement<?> ddlStatement = statementParser.parseSingleStatement(ddlText);
    final ConfiguredStatement<?> configuredStatement =
        ConfiguredStatement
            .of(ddlStatement, SessionConfig.of(originalConfig, emptyMap()));
    ksqlEngine.execute(serviceContext, configuredStatement);

    when(mockQueryMetadata.getQueryId()).thenReturn(mock(QueryId.class));

    final KsqlPlan plan = Mockito.mock(KsqlPlan.class);
    final Command csasCommand = new Command(
        statementText,
        emptyMap(),
        originalConfig.getAllConfigPropsWithSecretsObfuscated(),
        Optional.of(plan)
    );
    final CommandId csasCommandId = new CommandId(
        CommandId.Type.STREAM,
        "_CSASGen",
        CommandId.Action.CREATE);

    when(mockEngine.execute(eq(serviceContext), eqConfiguredPlan(plan), eq(false)))
        .thenReturn(ExecuteResult.of(mockQueryMetadata));

    // When:
    handleStatement(statementExecutorWithMocks, csasCommand, csasCommandId, Optional.empty(), 1);

    // Then:
    verify(mockQueryMetadata, times(1)).start();
  }