/**
 * Copyright 2017 Confluent Inc.
 **/

package io.confluent.ksql.cli;


import io.confluent.ksql.metastore.MetaStore;
import io.confluent.ksql.util.KafkaTopicClientImpl;
import io.confluent.ksql.util.KsqlConfig;
import io.confluent.ksql.util.Pair;
import io.confluent.ksql.util.PersistentQueryMetadata;
import io.confluent.ksql.util.QueryMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;

import io.confluent.ksql.KsqlEngine;
import io.confluent.ksql.parser.tree.Statement;

public class StandaloneExecutor {

  private static final Logger log = LoggerFactory.getLogger(StandaloneExecutor.class);

  KsqlEngine ksqlEngine;

  public StandaloneExecutor(Map streamProperties) {
    KsqlConfig ksqlConfig = new KsqlConfig(streamProperties);
    ksqlEngine = new KsqlEngine(ksqlConfig, new KafkaTopicClientImpl(ksqlConfig));
  }

  public void executeStatements(String queries) throws Exception {
    MetaStore tempMetaStore = ksqlEngine.getMetaStore().clone();
    List<Pair<String, Statement>> queryList = ksqlEngine.parseQueries(queries,
                                                                      Collections.emptyMap(),
                                                                      tempMetaStore);
    List<QueryMetadata> queryMetadataList = ksqlEngine.planQueries(
        false, queryList, Collections.emptyMap(), tempMetaStore);
    for (QueryMetadata queryMetadata: queryMetadataList) {
      if (queryMetadata instanceof PersistentQueryMetadata) {
        PersistentQueryMetadata persistentQueryMetadata = (PersistentQueryMetadata) queryMetadata;
        persistentQueryMetadata.getKafkaStreams().start();
      } else {
        System.err.println("Ignoring statemenst: " + queryMetadata.getStatementString());
        System.err.println("Only CREATE statements can run in KSQL embedded mode.");
        log.warn("Ignoring statemenst: {}", queryMetadata.getStatementString());
        log.warn("Only CREATE statements can run in KSQL embedded mode.");
      }
    }
  }

}

@@ -433,16 +433,42 @@
 ctions;%0A
+import java.util.HashMap;%0A
 import j
@@ -1427,36 +1427,29 @@
 ryList, 
-Collections.empty
+new Hash
 Map
+%3C%3E
 (), temp

SET is supported in application mode.