### Question:
KsqlContext { public void sql(String sql) throws Exception { List<QueryMetadata> queryMetadataList = ksqlEngine.buildMultipleQueries(sql, Collections .emptyMap()); for (QueryMetadata queryMetadata : queryMetadataList) { if (queryMetadata instanceof PersistentQueryMetadata) { PersistentQueryMetadata persistentQueryMetadata = (PersistentQueryMetadata) queryMetadata; persistentQueryMetadata.getKafkaStreams().start(); ksqlEngine.getPersistentQueries() .put(persistentQueryMetadata.getQueryId(), persistentQueryMetadata); } else { System.err.println("Ignoring statemenst: " + sql); System.err.println("Only CREATE statements can run in KSQL embedded mode."); log.warn("Ignoring statemenst: {}", sql); log.warn("Only CREATE statements can run in KSQL embedded mode."); } } } KsqlContext(
      final AdminClient adminClient,
      final KafkaTopicClient topicClient,
      final KsqlEngine ksqlEngine
  ); static KsqlContext create(KsqlConfig ksqlConfig); static KsqlContext create(
      KsqlConfig ksqlConfig,
      SchemaRegistryClient schemaRegistryClient
  ); MetaStore getMetaStore(); void sql(String sql); Set<QueryMetadata> getRunningQueries(); void close();  }### Answer:
@Test public void shouldRunSimpleStatements() throws Exception { AdminClient adminClient = mock(AdminClient.class); KafkaTopicClient kafkaTopicClient = mock(KafkaTopicClientImpl.class); KsqlEngine ksqlEngine = mock(KsqlEngine.class); Map<QueryId, PersistentQueryMetadata> liveQueryMap = new HashMap<>(); KsqlContext ksqlContext = new KsqlContext(adminClient, kafkaTopicClient, ksqlEngine); expect(ksqlEngine.buildMultipleQueries(statement1, Collections.emptyMap())) .andReturn (Collections.emptyList()); expect(ksqlEngine.buildMultipleQueries(statement2, Collections.emptyMap())) .andReturn(getQueryMetadata(new QueryId("CSAS_BIGORDERS"), DataSource.DataSourceType.KSTREAM)); expect(ksqlEngine.getPersistentQueries()).andReturn(liveQueryMap); replay(ksqlEngine); ksqlContext.sql(statement1); ksqlContext.sql(statement2); verify(ksqlEngine); }