Merge pull request #29 from RedPillAnalytics/develop

pipelineExecute is working flawlessly.

src/ksqlPipelinesTest/groovy/ExecuteTest.groovy

@@ -33,6 +33,7 @@ class ExecuteTest extends Specification {
                |plugins {
                |  id 'com.redpillanalytics.gradle-confluent'
                |}
+               confluent.pipelineEndpoint = 'http://localhost:8088'
         """.stripMargin())
    }
 
@@ -79,4 +80,70 @@ class ExecuteTest extends Specification {
       result.task(":${taskName}").outcome.name() != 'FAILED'
 
    }
+
+   def "Execute :pipelineExecute task with custom directory"() {
+
+      given:
+      taskName = 'pipelineExecute'
+      result = executeSingleTask(taskName, ['--pipeline-dir=src/main/pipeline/01-clickstream','-Si','--rerun-tasks'])
+
+      expect:
+      result.task(":${taskName}").outcome.name() != 'FAILED'
+
+   }
+
+   def "Execute :pipelineExecute task with --no-drop"() {
+
+      given:
+      taskName = 'pipelineExecute'
+      result = executeSingleTask(taskName, ['--no-drop','-Si','--rerun-tasks'])
+
+      expect:
+      result.task(":${taskName}").outcome.name() != 'FAILED'
+      !result.output.toLowerCase().contains('drop table')
+
+   }
+
+   def "Execute :pipelineExecute task with --no-terminate"() {
+
+      given:
+      taskName = 'pipelineExecute'
+      result = executeSingleTask(taskName, ['--no-terminate','-Si','--rerun-tasks'])
+
+      expect:
+      result.task(":${taskName}").outcome.name() != 'FAILED'
+      !result.output.toLowerCase().contains('terminate')
+
+   }
+
+   def "Execute :pipelineExecute task with --no-create"() {
+
+      given:
+      taskName = 'pipelineExecute'
+      result = executeSingleTask(taskName, ['--no-create','-Si','--rerun-tasks'])
+
+      expect:
+      result.task(":${taskName}").outcome.name() != 'FAILED'
+      !result.output.toLowerCase().contains('create table')
+      !result.output.toLowerCase().contains('insert into')
+
+   }
+
+   def "Execute :pipelineExecute task with custom REST endpoint"() {
+
+      given:
+      buildFile.write("""
+               |plugins {
+               |  id 'com.redpillanalytics.gradle-confluent'
+               |}
+               confluent.pipelineEndpoint = 'http://nothing:8088'
+        """.stripMargin())
+
+      taskName = 'pipelineExecute'
+      result = executeSingleTask(taskName, ["-Pconfluent.pipelineEndpoint=http://localhost:8088",'-Si','--rerun-tasks'])
+
+      expect:
+      result.task(":${taskName}").outcome.name() != 'FAILED'
+
+   }
 }

src/main/groovy/com/redpillanalytics/KsqlRest.groovy

@@ -119,7 +119,7 @@ class KsqlRest {
 
             // now terminate with extreme prejudice
             queries.each { queryId ->
-               execKsql("TERMINATE ${queryId}")
+               execKsql("TERMINATE ${queryId}", properties)
             }
 
             // now drop the table again

src/main/groovy/com/redpillanalytics/gradle/ConfluentPluginExtension.groovy

@@ -73,7 +73,7 @@ class ConfluentPluginExtension {
    /**
     * RESTful endpoint for the KSQL Server.
     */
-   String pipelineEndpoint= 'http://localhost:8088'
+   String pipelineEndpoint = 'http://localhost:8088'
 
    /**
     * The pattern used for matching the pipeline deployment artifact.

src/main/groovy/com/redpillanalytics/gradle/tasks/PipelineExecuteTask.groovy

@@ -1,14 +1,43 @@
 package com.redpillanalytics.gradle.tasks
 
 import groovy.util.logging.Slf4j
+import org.gradle.api.tasks.Input
 import org.gradle.api.tasks.TaskAction
+import org.gradle.api.tasks.options.Option
 
 /**
  * Use the KSQL RESTful API to execute all pipelines in a particular directory. Note: not functioning yet.
  */
 @Slf4j
 class PipelineExecuteTask extends PipelineTask {
 
+   /**
+    * When defined, the DROPS script is not constructed in reverse order.
+    */
+   @Input
+   @Option(option = 'no-terminate',
+           description = 'When defined, DROP statements are not processed using a TERMINATE for all currently-running queries.'
+   )
+   boolean noTerminate
+
+   /**
+    * When defined, the DROPS script is not constructed in reverse order.
+    */
+   @Input
+   @Option(option = 'no-drop',
+           description = 'When defined, DROP statements are not processed.'
+   )
+   boolean noDrop
+
+   /**
+    * When defined, the DROPS script is not constructed in reverse order.
+    */
+   @Input
+   @Option(option = 'no-create',
+           description = 'When defined, CREATE statements are not processed.'
+   )
+   boolean noCreate
+
    /**
     * The main Gradle Task method.
     */
@@ -17,10 +46,10 @@ class PipelineExecuteTask extends PipelineTask {
 
       // first execute the DROP SQL statements
       // this also catches running statements and terminates them
-      ksqlRest.dropKsql(dropSql, [:])
+      if (!noDrop) ksqlRest.dropKsql(dropSql, [:], !noTerminate)
 
       // now create the pipelines
-      ksqlRest.execKsql(pipelineSql)
+      if (!noCreate) ksqlRest.execKsql(pipelineSql)
 
    }
 }