lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

shutdown all KafkaStream apps if one crashes

Browse source
This commit is contained in:
Adam 2022-03-12 01:39:19 +01:00
parent 7a46ea7b75
commit 19bfe3a232
3 changed files with 8 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

4
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt
View file

@ -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(

5
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/StreamsExceptionHandler.kt
View file

@ -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
}
}