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

max task idle ms - try fixing ktable-ktable join...

Browse source
This commit is contained in:
Adam 2022-03-13 23:58:41 +01:00
parent 8c22329b10
commit 0ebf492b42
1 changed files with 2 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -9,6 +9,7 @@ import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream
import java.time.Duration import java.time.Duration
import kotlin.coroutines.CoroutineContext import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.cancellation.CancellationException import kotlin.coroutines.cancellation.CancellationException
import kotlin.time.Duration.Companion.minutes
import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
@ -80,6 +81,7 @@ internal class KafkatorioTopology(
val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as? String
props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId) props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
props.setProperty(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, "${1.minutes.inWholeMilliseconds}")
val streams = KafkaStreams(topology, props) val streams = KafkaStreams(topology, props)
streams.setUncaughtExceptionHandler(StreamsExceptionHandler { streams.setUncaughtExceptionHandler(StreamsExceptionHandler {