From 19bfe3a232aee89f74c89bf0b6b0bbb74843d22e Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Sat, 12 Mar 2022 01:39:19 +0100
Subject: [PATCH] shutdown all KafkaStream apps if one crashes

---
 .../dev/adamko/kafkatorio/events/schema/factorioIds.kt       | 2 +-
 .../dev/adamko/kafkatorio/processor/KafkatorioTopology.kt    | 4 +++-
 .../adamko/kafkatorio/processor/StreamsExceptionHandler.kt   | 5 ++++-
 3 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioIds.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioIds.kt
index 4dcdb76..e8a9e90 100644
--- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioIds.kt
+++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioIds.kt
@@ -31,7 +31,7 @@ value class PlayerIndex(private val id: UInt) {
 }
 
 
-/** This is universally unique for every entity that has one, for the lifetime of a whole game. */
+/** > This is universally unique for every entity that has one, for the lifetime of a whole game. */
 @Serializable
 @JvmInline
 value class UnitNumber(private val id: UInt) {
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
index 1e8dd10..3d35424 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
@@ -156,7 +156,9 @@ internal class KafkatorioTopology(
     props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
 
     val streams = KafkaStreams(topology, props)
-    streams.setUncaughtExceptionHandler(StreamsExceptionHandler())
+    streams.setUncaughtExceptionHandler(StreamsExceptionHandler {
+      coroutineContext.job.cancel(CancellationException(cause = it))
+    })
 
     val description: TopologyDescription = topology.describe()
     println(
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt
index e188560..18ab16a 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt
@@ -3,9 +3,12 @@ package dev.adamko.kafkatorio.processor
 import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler
 import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse
 
-class StreamsExceptionHandler : StreamsUncaughtExceptionHandler {
+class StreamsExceptionHandler(
+  val shutdown: (e: Throwable) -> Unit,
+) : StreamsUncaughtExceptionHandler {
   override fun handle(exception: Throwable): StreamThreadExceptionResponse {
     exception.printStackTrace()
+    shutdown(exception)
     return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION
   }
 }