improve kafka consumer group gradle task

buildSrc/src/main/kotlin/dev/adamko/kafkatorio/task/KafkaConsumersResetTask.kt

@@ -2,60 +2,76 @@ package dev.adamko.kafkatorio.task
 
 import dev.adamko.kafkatorio.gradle.execCapture
 import org.gradle.api.DefaultTask
-import org.gradle.api.file.DirectoryProperty
 import org.gradle.api.logging.LogLevel
-import org.gradle.api.tasks.Destroys
 import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.Internal
 import org.gradle.api.tasks.TaskAction
 import org.intellij.lang.annotations.Language
 import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
 
-abstract class KafkaConsumersResetTask : DefaultTask() {
+abstract class KafkaConsumerGroupsTask : DefaultTask() {
 
   @get:Input
   val bootstrapServer: String = "localhost:9092"
   @get:Input
-  val kafkaContainerName: String = "kafka"
+  val kafkaDockerContainerName: String = "kafka"
-  @get:Destroys
-  abstract val kafkaStateDir: DirectoryProperty
 
+  @get:Internal
+  protected val bootstrapArg = "--bootstrap-server $bootstrapServer"
 
-  @TaskAction
+  abstract class ResetAll : KafkaConsumerGroupsTask() {
-  fun resetKafkaConsumers() {
-    val bootstrapArg = "--bootstrap-server $bootstrapServer"
 
-    val allGroups = dockerExec(
+    @get:Input
+    @Language("Shell Script")
+    val resetCommand: String =
       """
-        /kafka/bin/kafka-consumer-groups.sh --list $bootstrapArg
+        /kafka/bin/kafka-consumer-groups.sh --execute --reset-offsets --all-topics --to-earliest $bootstrapArg
       """.trimIndent()
-    ).lines()
-      .filter { it.trim().isNotBlank() }
 
-    allGroups
+    @TaskAction
-      .forEach { group ->
+    fun reset() {
+      getAllGroups().forEach { group ->
         logger.lifecycle("Resetting group $group\n---")
-        dockerExec(
+        dockerExec("$resetCommand --group $group")
-          """
-            /kafka/bin/kafka-consumer-groups.sh --group $group --execute --reset-offsets --all-topics --to-earliest $bootstrapArg
-          """.trimIndent()
-        )
         logger.lifecycle("---")
       }
-
-    project.delete {
-      delete(kafkaStateDir)
     }
   }
 
+  abstract class DeleteAll : KafkaConsumerGroupsTask() {
+
+    @get:Input
+    @Language("Shell Script")
+    val deleteCommand: String =
+      """
+        /kafka/bin/kafka-consumer-groups.sh --delete $bootstrapArg
+      """.trimIndent()
+
+    @TaskAction
+    fun reset() {
+      val groups = getAllGroups().joinToString(" ") { group -> "--group $group" }
+      dockerExec("$deleteCommand $groups")
+    }
+  }
+
+  @Internal
+  protected fun getAllGroups(): List<String> = dockerExec(
+    """
+      /kafka/bin/kafka-consumer-groups.sh --list $bootstrapArg
+    """.trimIndent()
+  ).lines()
+    .filter { it.trim().isNotBlank() }
+
+
 //  fun reset(applicationId: String) = """
 //      /kafka/bin/kafka-streams-application-reset.sh --application-id $applicationId --force --to-earliest
 //    """.trimIndent()
 
-  private fun dockerExec(
+  protected fun dockerExec(
     @Language("Shell Script")
     cmd: String
   ): String = project.execCapture {
     logging.captureStandardOutput(LogLevel.LIFECYCLE)
-    commandLine = parseSpaceSeparatedArgs(""" docker exec $kafkaContainerName $cmd """)
+    commandLine = parseSpaceSeparatedArgs(""" docker exec $kafkaDockerContainerName $cmd """)
   }
 }

modules/events-server/build.gradle.kts

@@ -1,4 +1,4 @@
-import dev.adamko.kafkatorio.task.KafkaConsumersResetTask
+import dev.adamko.kafkatorio.task.KafkaConsumerGroupsTask
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
 plugins {
@@ -63,9 +63,14 @@ tasks.withType<KotlinCompile> {
 }
 
 
-val kafkaConsumersReset by tasks.registering(KafkaConsumersResetTask::class) {
+val kafkaStateDirDelete by tasks.registering(Delete::class) {
+  group = project.name
+  mustRunAfter(kafkaConsumersDelete)
+  delete(layout.buildDirectory.dir("kafka-state"))
+}
+
+val kafkaConsumersDelete by tasks.registering(KafkaConsumerGroupsTask.DeleteAll::class) {
   group = project.name
-  kafkaStateDir.set(project.layout.buildDirectory.dir("kafka-state"))
 }
 
 //tasks.run.configure {