Kafka Streams working again?

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt

@@ -14,7 +14,6 @@ import kotlin.reflect.KClass
 
 private const val DOMAIN = "kafkatorio"
 
-
 const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log"
 
 const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped"

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/groupMapChunks.kt

@@ -13,9 +13,9 @@ import dev.adamko.kafkatorio.schema.common.toMapChunkPosition
 import dev.adamko.kafkatorio.schema.packets.MapChunkUpdate
 import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
 import dev.adamko.kotka.extensions.groupedAs
-import dev.adamko.kotka.extensions.materializedAs
 import dev.adamko.kotka.extensions.materializedWith
 import dev.adamko.kotka.extensions.producedAs
+import dev.adamko.kotka.extensions.repartitionedAs
 import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.flatMap
 import dev.adamko.kotka.extensions.streams.mapValues
@@ -66,21 +66,22 @@ fun groupMapChunks(builder: StreamsBuilder): Topology {
       tileProtoColourDict,
     )
 
+  val pid = "grouped-map-chunk-tiles.output"
   groupedMapChunkTiles
-    .toStream("stream-grouped-map-tiles")
-    .filter("stream-grouped-map-tiles.filter-not-null") { _, v ->
+    .toStream("$pid.to-stream")
+    .filter("$pid.filter-tiles-not-empty") { _, v ->
       !v?.map.isNullOrEmpty()
     }
-    .mapValues("stream-grouped-map-tiles.map-not-null") { _, v ->
+    .mapValues("$pid.map-not-null") { _, v ->
       requireNotNull(v)
     }
-    .peek("stream-grouped-map-tiles.peek") { _, v ->
+    .peek("$pid.print-group-result") { _, v ->
       println("Grouping map tiles result: ${v.chunkId} / ${v.map.size}")
     }
     .to(
       TOPIC_GROUPED_MAP_CHUNKS_STATE,
       producedAs(
-        "produce.grouped-map-chunks",
+        "$pid.grouped-map-chunks",
         kxsBinary.serde<ServerMapChunkId>(),
         kxsBinary.serde<ServerMapChunkTiles<ColourHex>>()
       )
@@ -102,9 +103,12 @@ private fun groupTilesIntoChunksWithColours(
     }.mapValues("map-chunk-update-packets.convert-to-map-tiles") { _, packet: MapChunkUpdate ->
       val mapTiles = packet.tileDictionary?.toMapTileList() ?: emptyList()
       MapTiles(packet.key.surfaceIndex, mapTiles)
-    }.filter { _, value ->
+    }.filter("map-chunk-update-packets.filter-out.no-tiles") { _, value ->
       value.tiles.isNotEmpty()
     }
+//      .peek { _, value ->
+//      println("[${System.currentTimeMillis()}] mapTilesStream: surface:${value.surfaceIndex}, tile count: ${value.tiles.size}")
+//    }
 
 
   val chunkedTilesTable: KTable<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>> =
@@ -115,8 +119,6 @@ private fun groupTilesIntoChunksWithColours(
         val standardChunkSize = ChunkSize.MAX
 
         val flatMappedChunkTiles: List<Pair<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>> =
-          //        println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles")
-
           mapTiles
             .tiles
             .groupBy { tile ->
@@ -145,6 +147,15 @@ private fun groupTilesIntoChunksWithColours(
       .filter("server-map-data.tiles.filter-not-empty") { _, value ->
         value.map.isNotEmpty()
       }
+      .repartition(
+        repartitionedAs(
+          "server-map-data.tiles.pre-table-repartition",
+          kxsBinary.serde(),
+          kxsBinary.serde(),
+          // force, otherwise KTable-KTable FK join doesn't work
+          numberOfPartitions = 1,
+        )
+      )
       .groupByKey(
         groupedAs(
           "server-map-data.tiles.group",
@@ -155,24 +166,35 @@ private fun groupTilesIntoChunksWithColours(
       .reduce(
         "server-map-data.tiles.reduce",
         materializedWith(
+//        materializedAs(
+//          "server-map-data.tiles.reduce.store",
           kxsBinary.serde<ServerMapChunkId>(),
           kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
         )
-      ) { chunkTiles, other -> chunkTiles + other }
-
+      ) { chunkTiles, other ->
+//        println(
+//          "[${System.currentTimeMillis()}] reducing tiles " +
+//              "chunkTiles:${chunkTiles.chunkId.chunkPosition}, ${chunkTiles.chunkId.chunkSize}, ${chunkTiles.map.size} " +
+//              "other:${other.chunkId.chunkPosition}, ${other.chunkId.chunkSize}, ${other.map.size}"
+//        )
+        chunkTiles + other
+      }
 
   val colourisedChunkTable: KTable<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
     chunkedTilesTable
       .join(
         other = tileProtoColourDict,
         tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"),
-        materialized = materializedAs(
-          "server-map-data.join-tiles-with-prototypes.store",
+        materialized = materializedWith(
+//          "server-map-data.join-tiles-with-prototypes.store",
           kxsBinary.serde<ServerMapChunkId>(),
           kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
         ),
-        foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> -> chunkTiles.chunkId.serverId }
+        foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> ->
+          chunkTiles.chunkId.serverId
+        }
       ) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
+//        println("joining tiles:${tiles.map.size} with colourDict:${colourDict.map.size}")
 
         val missingProtos = mutableSetOf<TileProtoHashCode>()
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/packetDataStreams.kt

@@ -5,6 +5,7 @@ import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
 import dev.adamko.kotka.extensions.consumedAs
+import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.kxs.serde
 import org.apache.kafka.streams.StreamsBuilder
@@ -16,12 +17,14 @@ inline fun <reified T : KafkatorioPacketData> StreamsBuilder.streamPacketData():
   val simpleName = T::class.simpleName!!
   val topicName = T::class.topicName
 
+  val taskName = "consume.${simpleName}"
+
   return stream<FactorioServerId, KafkatorioPacket>(
     topicName,
-    consumedAs("consume.${simpleName}", jsonMapper.serde(), jsonMapper.serde())
-  ).filter { key: FactorioServerId?, value: KafkatorioPacket? ->
+    consumedAs("$taskName.input", jsonMapper.serde(), jsonMapper.serde())
+  ).filter("$taskName.filter-is-instance") { key: FactorioServerId?, value: KafkatorioPacket? ->
     key != null && value?.data is T
-  }.mapValues("consume.map-values.${simpleName}") { _: FactorioServerId, packet: KafkatorioPacket ->
+  }.mapValues("$taskName.map-values") { _: FactorioServerId, packet: KafkatorioPacket ->
     packet.data as T
   }
 }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt

@@ -4,9 +4,10 @@ import dev.adamko.kafkatorio.processor.serdes.kxsBinary
 import dev.adamko.kafkatorio.schema.common.ColourHex
 import dev.adamko.kafkatorio.schema.common.MapTile
 import dev.adamko.kafkatorio.schema.common.toHex
-import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype
 import dev.adamko.kafkatorio.schema.packets.PrototypesUpdate
+import dev.adamko.kafkatorio.schema.prototypes.FactorioPrototype
 import dev.adamko.kotka.extensions.materializedAs
+import dev.adamko.kotka.extensions.repartitionedAs
 import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.extensions.streams.toTable
@@ -34,51 +35,35 @@ value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
 fun tileProtoColourDictionary(
   factorioServerPacketStream: KStream<FactorioServerId, PrototypesUpdate>
 ): KTable<FactorioServerId, TileColourDict> {
+  val pid = "tileProtoColourDictionary"
 
-  return factorioServerPacketStream.mapValues("server-map-data.tile-prototypes.map-values") { _, protoPacket: PrototypesUpdate ->
+  return factorioServerPacketStream.mapValues(
+    "$pid.map-values"
+  ) { _, protoPacket: PrototypesUpdate ->
     val map = protoPacket
       .prototypes
       .filterIsInstance<FactorioPrototype.MapTile>()
       .associate { TileProtoHashCode(it) to it.mapColour.toHex() }
 
     TileColourDict(map)
-  }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
+  }.filter("$pid.filterMapTileProtos") { _, dict: TileColourDict ->
     dict.map.isNotEmpty()
   }.peek { serverId, dict: TileColourDict ->
-    println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
-  }.toTable(
-    "server-map-data.tile-prototypes",
+    println("$pid: server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
+  }.repartition(
+    repartitionedAs(
+      "$pid.pre-table-repartition",
+      kxsBinary.serde(),
+      kxsBinary.serde(),
+      // force, otherwise KTable-KTable FK join doesn't work
+      numberOfPartitions = 1,
+    )
+  ).toTable(
+    "$pid.create-table",
     materializedAs(
-      "server-map-data.tile-prototypes.store",
+      "$pid.output-store",
       kxsBinary.serde(),
       kxsBinary.serde(),
     )
   )
 }
-
-// get all the prototypes
-//  return factorioServerPacketStream
-//    .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet: KafkatorioPacket ->
-//      val map = when (packet) {
-//        is FactorioPrototype -> packet
-//          .prototypes
-//          .filterIsInstance<MapTilePrototype>()
-//          .associate { TileProtoHashCode(it) to it.mapColour.toHex() }
-//        else                 -> mapOf()
-//      }
-//      TileColourDict(map)
-//    }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
-//      dict.map.isNotEmpty()
-//    }.peek { serverId, dict: TileColourDict ->
-//      println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
-//    }
-//    .toTable(
-//      "server-map-data.tile-prototypes",
-//      materializedAs(
-//        "server-map-data.tile-prototypes.store",
-//        kxsBinary.serde(),
-//        kxsBinary.serde(),
-//      )
-//    )
-//
-//}