create KafkaConsumers reset task

buildSrc/src/main/kotlin/dev/adamko/kafkatorio/gradle/extensions.kt

@@ -12,6 +12,7 @@ import org.gradle.api.provider.MapProperty
 import org.gradle.api.specs.Spec
 import org.gradle.kotlin.dsl.invoke
 import org.gradle.plugins.ide.idea.model.IdeaModule
+import org.gradle.process.ExecSpec
 
 
 operator fun <T> Spec<T>.not(): Spec<T> = Spec<T> { !this(it) }
@@ -71,3 +72,17 @@ fun IdeaModule.excludeGeneratedGradleDsl(layout: ProjectLayout) {
     )
   )
 }
+
+
+// https://stackoverflow.com/a/70317110/4161471
+fun Project.execCapture(spec: ExecSpec.() -> Unit): String {
+  return ByteArrayOutputStream().use { outputStream ->
+    exec {
+      this.spec()
+      this.standardOutput = outputStream
+    }
+    val output = outputStream.toString().trim()
+    logger.lifecycle(output)
+    output
+  }
+}

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

@@ -0,0 +1,61 @@
+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.TaskAction
+import org.intellij.lang.annotations.Language
+import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
+
+abstract class KafkaConsumersResetTask : DefaultTask() {
+
+  @get:Input
+  val bootstrapServer: String = "localhost:9092"
+  @get:Input
+  val kafkaContainerName: String = "kafka"
+  @get:Destroys
+  abstract val kafkaStateDir: DirectoryProperty
+
+
+  @TaskAction
+  fun resetKafkaConsumers() {
+    val bootstrapArg = "--bootstrap-server $bootstrapServer"
+
+    val allGroups = dockerExec(
+      """
+        /kafka/bin/kafka-consumer-groups.sh --list $bootstrapArg
+      """.trimIndent()
+    ).lines()
+      .filter { it.trim().isNotBlank() }
+
+    allGroups
+      .forEach { group ->
+        logger.lifecycle("Resetting group $group\n---")
+        dockerExec(
+          """
+            /kafka/bin/kafka-consumer-groups.sh --group $group --execute --reset-offsets --all-topics --to-earliest $bootstrapArg
+          """.trimIndent()
+        )
+        logger.lifecycle("---")
+      }
+
+    project.delete {
+      delete(kafkaStateDir)
+    }
+  }
+
+//  fun reset(applicationId: String) = """
+//      /kafka/bin/kafka-streams-application-reset.sh --application-id $applicationId --force --to-earliest
+//    """.trimIndent()
+
+  private fun dockerExec(
+    @Language("Shell Script")
+    cmd: String
+  ): String = project.execCapture {
+    logging.captureStandardOutput(LogLevel.LIFECYCLE)
+    commandLine = parseSpaceSeparatedArgs(""" docker exec $kafkaContainerName $cmd """)
+  }
+}

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

@@ -8,6 +8,7 @@ import org.gradle.api.tasks.OutputDirectory
 import org.gradle.api.tasks.PathSensitive
 import org.gradle.api.tasks.PathSensitivity
 import org.gradle.api.tasks.SkipWhenEmpty
+import org.gradle.work.NormalizeLineEndings
 import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
 
 abstract class TypescriptToLuaTask : NpmTask() {
@@ -15,6 +16,8 @@ abstract class TypescriptToLuaTask : NpmTask() {
   @get:InputDirectory
   @get:SkipWhenEmpty
   @get:PathSensitive(PathSensitivity.RELATIVE)
+  @Suppress("UnstableApiUsage")
+  @get:NormalizeLineEndings
   abstract val sourceFiles: DirectoryProperty
 
   @get:OutputDirectory

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

@@ -7,8 +7,10 @@ import org.gradle.api.tasks.Input
 import org.gradle.api.tasks.OutputFile
 
 abstract class UpdatePackageJson : DefaultTask() {
+
   @get:Input
   abstract val propertiesToCheck: MapProperty<String, String>
+
   @get:OutputFile
   abstract val packageJsonFile: RegularFileProperty
 

modules/events-server/build.gradle.kts

@@ -1,5 +1,5 @@
+import dev.adamko.kafkatorio.task.KafkaConsumersResetTask
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
-import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
 
 plugins {
   dev.adamko.kafkatorio.lang.`kotlin-jvm`
@@ -58,39 +58,14 @@ tasks.withType<KotlinCompile> {
   kotlinOptions.freeCompilerArgs += listOf(
     "-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi",
     "-opt-in=kotlinx.coroutines.FlowPreview",
+    "-opt-in=kotlinx.serialization.ExperimentalSerializationApi",
   )
 }
 
-val kafkatorioEventsServerKafkaForceReset by tasks.registering {
+
+val kafkaConsumersReset by tasks.registering(KafkaConsumersResetTask::class) {
   group = project.name
-
+  kafkaStateDir.set(project.layout.buildDirectory.dir("kafka-state"))
-  fun reset(applicationId: String) = """
-    /kafka/bin/kafka-streams-application-reset.sh --application-id $applicationId --force
-  """.trimIndent()
-
-  doLast {
-
-    exec {
-      val cmd = reset("kafkatorio-events-processor.saveTiles")
-      logging.captureStandardOutput(LogLevel.LIFECYCLE)
-      commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """)
-    }
-    exec {
-      val cmd = reset("kafkatorio-events-processor.splitPackets")
-      logging.captureStandardOutput(LogLevel.LIFECYCLE)
-      commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd  --input-topics factorio-server-log  """)
-    }
-    exec {
-      val cmd = reset("kafkatorio-events-processor.playerUpdates")
-      logging.captureStandardOutput(LogLevel.LIFECYCLE)
-      commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """)
-    }
-    exec {
-      val cmd = reset("kafkatorio-events-processor.groupTilesMapChunks")
-      logging.captureStandardOutput(LogLevel.LIFECYCLE)
-      commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """)
-    }
-  }
 }
 
 //tasks.run.configure {