### Question:
KsqlEngine implements Closeable { public List<QueryMetadata> buildMultipleQueries( final String queriesString, final Map<String, Object> overriddenProperties ) throws Exception { for (String property : overriddenProperties.keySet()) { if (IMMUTABLE_PROPERTIES.contains(property)) { throw new IllegalArgumentException( String.format("Cannot override property '%s'", property) ); } } MetaStore tempMetaStore = metaStore.clone(); List<Pair<String, Statement>> queries = parseQueries( queriesString, overriddenProperties, tempMetaStore ); return planQueries(queries, overriddenProperties, tempMetaStore); } 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 shouldFailDDLStatementIfTopicDoesNotExist() { String ddlStatement = "CREATE STREAM S1_NOTEXIST (COL1 BIGINT, COL2 VARCHAR) " + "WITH (KAFKA_TOPIC = 'S1_NOTEXIST', VALUE_FORMAT = 'JSON');"; try { List<QueryMetadata> queries = ksqlEngine.buildMultipleQueries(ddlStatement.toString(), Collections.emptyMap()); Assert.fail(); } catch (Exception e) { assertThat(e.getMessage(), equalTo("Kafka topic does not exist: S1_NOTEXIST")); } }
@Test public void shouldEnforceTopicExistenceCorrectly() throws Exception { topicClient.createTopic("s1_topic", 1, (short) 1); 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<QueryMetadata> queries = ksqlEngine.buildMultipleQueries(runScriptContent.toString(), Collections.emptyMap()); Assert.assertTrue(topicClient.isTopicExists("T1")); }