---
up:
  - name: create source connector dogs
    exec: |
      CREATE SOURCE CONNECTOR DOGS WITH ('connector.class'='io.mdrogalis.voluble.VolubleSourceConnector',
      'key.converter'='org.apache.kafka.connect.storage.StringConverter',
      'value.converter'='org.apache.kafka.connect.json.JsonConverter',
      'value.converter.schemas.enable'='false',
      'genkp.dogs.with'='#{Internet.uuid}',
      'genv.dogs.name.with'='#{Dog.name}',
      'genv.dogs.dogsize.with'='#{Dog.size}',
      'genv.dogs.age.with'='#{Dog.age}',
      'topic.dogs.throttle.ms'=1000 
      );
  - name: create the dogs stream
    exec: |
      CREATE STREAM IF NOT EXISTS DOGS (ID STRING KEY,NAME STRING,DOGSIZE STRING, AGE STRING) 
      WITH (KAFKA_TOPIC='dogs', 
      VALUE_FORMAT='JSON', PARTITIONS=1);
  - name: create the DOGS_BY_SIZE table
    exec: |
      CREATE TABLE IF NOT EXISTS DOGS_BY_SIZE AS 
      SELECT DOGSIZE AS DOG_SIZE, COUNT(*) AS DOGS_CT 
      FROM DOGS WINDOW TUMBLING (SIZE 15 MINUTE) 
      GROUP BY DOGSIZE;
down:
  - name: drop table DOGS_BY_SIZE
    exec: DROP TABLE IF EXISTS DOGS_BY_SIZE;
  - name: delete stream DOGS
    exec: DROP STREAM IF EXISTS DOGS;
  - name: drop connector DOGS
    exec: DROP CONNECTOR IF EXISTS DOGS;