### Question:
KsqlEngine implements Closeable { List<Pair<String, Statement>> parseQueries( final String queriesString, final Map<String, Object> overriddenProperties, final MetaStore tempMetaStore ) { try { MetaStore tempMetaStoreForParser = tempMetaStore.clone(); KsqlParser ksqlParser = new KsqlParser(); List<SqlBaseParser.SingleStatementContext> parsedStatements = ksqlParser.getStatements(queriesString); List<Pair<String, Statement>> queryList = new ArrayList<>(); for (SqlBaseParser.SingleStatementContext singleStatementContext : parsedStatements) { Pair<Statement, DataSourceExtractor> statementInfo = ksqlParser.prepareStatement( singleStatementContext, tempMetaStoreForParser ); Statement statement = statementInfo.getLeft(); Pair<String, Statement> queryPair = buildSingleQueryAst( statement, getStatementString(singleStatementContext), tempMetaStore, tempMetaStoreForParser, overriddenProperties ); if (queryPair != null) { queryList.add(queryPair); } } return queryList; } catch (Exception e) { throw new ParseFailedException("Exception while processing statements :" + e.getMessage(), e); } } KsqlEngine(final KsqlConfig ksqlConfig, final KafkaTopicClient topicClient);  KsqlEngine(
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient topicClient,
      final SchemaRegistryClient schemaRegistryClient,
      final MetaStore metaStore
  ); List<QueryMetadata> buildMultipleQueries(
      final String queriesString,
      final Map<String, Object> overriddenProperties
  ); QueryMetadata getQueryExecutionPlan(final Query query); static String getStatementString(
      final SqlBaseParser.SingleStatementContext singleStatementContext
  ); List<Statement> getStatements(final String sqlString); Query addInto(
      final Query query, final QuerySpecification querySpecification,
      final String intoName,
      final Map<String, Expression> intoProperties,
      final Optional<Expression> partitionByExpression
  ); Set<QueryMetadata> getLivePersistentQueries(); MetaStore getMetaStore(); FunctionRegistry getFunctionRegistry(); KafkaTopicClient getTopicClient(); DdlCommandExec getDdlCommandExec(); boolean terminateQuery(final QueryId queryId, final boolean closeStreams); Map<QueryId, PersistentQueryMetadata> getPersistentQueries(); static List<String> getImmutableProperties(); Map<String, Object> getKsqlConfigProperties(); KsqlConfig getKsqlConfig(); long numberOfLiveQueries(); long numberOfPersistentQueries(); @Override void close(); void removeTemporaryQuery(QueryMetadata queryMetadata); DdlCommandResult executeDdlStatement(
      String sqlExpression, final DdlStatement statement,
      final Map<String, Object> overriddenProperties
  ); SchemaRegistryClient getSchemaRegistryClient(); List<QueryMetadata> createQueries(final String queries);  }### Answer:
@Test public void shouldNotEnforceTopicExistanceWhileParsing() throws Exception { StringBuilder runScriptContent = new StringBuilder("CREATE STREAM S1 (COL1 BIGINT, COL2 VARCHAR) " + "WITH (KAFKA_TOPIC = 's1_topic', VALUE_FORMAT = 'JSON');\n"); runScriptContent.append("CREATE TABLE T1 AS SELECT COL1, count(*) FROM " + "S1 GROUP BY COL1;\n"); runScriptContent.append("CREATE STREAM S2 (C1 BIGINT, C2 BIGINT) " + "WITH (KAFKA_TOPIC = 'T1', VALUE_FORMAT = 'JSON');\n"); List<Pair<String, Statement>> parsedStatements = ksqlEngine.parseQueries( runScriptContent.toString(), Collections.emptyMap(), metaStore.clone()); assertThat(parsedStatements.size(), equalTo(3)); }