From 0ebf492b42d2b2da3181dff08fcace0a6bffc171 Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Sun, 13 Mar 2022 23:58:41 +0100
Subject: [PATCH] max task idle ms - try fixing ktable-ktable join...

---
 .../dev/adamko/kafkatorio/processor/KafkatorioTopology.kt       | 2 ++
 1 file changed, 2 insertions(+)

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