- bump kafka sdk version

- fix kafka admin so it doesn't try making topics always
- move some stuff to Kotka Streams
- trying to optimise and debounce with windows... but it didn't work
This commit is contained in:
Adam 2022-02-05 12:20:25 +01:00
parent f5855ba5bd
commit f108431033
11 changed files with 216 additions and 135 deletions

View file

@ -38,7 +38,8 @@ dependencies {
implementation("com.sksamuel.scrimage:scrimage-core:4.0.24")
val kotkaVersion = "0.0.8"
// val kotkaVersion = "0.0.11"
val kotkaVersion = "store-iterators-SNAPSHOT"
implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") {
isChanging = true
}
@ -49,6 +50,10 @@ dependencies {
}
//configurations.all {
// resolutionStrategy.cacheDynamicVersionsFor(1, TimeUnit.MINUTES)
//}
application {
mainClass.set("dev.adamko.kafkatorio.processor.EventProcessorKt")
}
@ -56,5 +61,6 @@ application {
tasks.withType<KotlinCompile> {
kotlinOptions.freeCompilerArgs += listOf(
"-opt-in=kotlinx.coroutines.ExperimentalCoroutinesApi",
"-opt-in=kotlinx.coroutines.FlowPreview",
)
}

View file

@ -28,11 +28,10 @@ object CreateTopics {
fun create() {
KafkatorioPacket.PacketType.values().forEach { packetType ->
val currentTopics = currentTopics().map { it.name() }
val currentTopics = currentTopics().map { it.name() }
val kafkatorioTopics = buildSet {
val kafkatorioTopics = buildSet {
KafkatorioPacket.PacketType.values().forEach { packetType ->
add(KafkatorioTopology.sourceTopic)
when (packetType) {
KafkatorioPacket.PacketType.EVENT ->
@ -44,30 +43,30 @@ object CreateTopics {
KafkatorioPacket.PacketType.PROTOTYPES ->
add("kafkatorio.${packetType.name}.all")
}
}.minus(currentTopics.toSet())
if (currentTopics.isNotEmpty()) {
logger.info("Creating ${kafkatorioTopics.size} topics")
val result = kafkaAdmin.createTopics { kafkatorioTopics }
// wait for all brokers to become aware of the created topics
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
result.values().forEach {
logger.info("Created topic: ${it.key} ${it.value.get()}")
}
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
logger.info("Listing topics")
kafkaAdmin.listTopics()
.listings()
.get()
.forEach {
logger.info("Topic: $it")
}
}
}.minus(currentTopics.toSet())
if (kafkatorioTopics.isNotEmpty()) {
logger.info("Creating ${kafkatorioTopics.size} topics")
val result = kafkaAdmin.createTopics { kafkatorioTopics }
// wait for all brokers to become aware of the created topics
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
result.values().forEach {
logger.info("Created topic: ${it.key} ${it.value.get()}")
}
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
logger.info("Listing topics")
kafkaAdmin.listTopics()
.listings()
.get()
.forEach {
logger.info("Topic: $it")
}
}
}

View file

@ -1,7 +1,7 @@
package dev.adamko.kafkatorio.processor
import dev.adamko.kafkatorio.events.schema.Colour
import dev.adamko.kafkatorio.events.schema.MapTile
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
import dev.adamko.kafkatorio.processor.topology.PrototypeName
import dev.adamko.kafkatorio.processor.topology.TileUpdateRecordKey
import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPixels
@ -9,9 +9,9 @@ import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition
import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles
import dev.adamko.kafkatorio.processor.topology.allMapTilesTable
import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer
import dev.adamko.kafkatorio.processor.topology.prototypesTable
import dev.adamko.kafkatorio.processor.topology.saveTileImages
import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog
import dev.adamko.kafkatorio.processor.topology.tilePrototypeColourTable
import java.time.Duration
import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.StreamsBuilder
@ -35,10 +35,10 @@ class KafkatorioTopology(
playerUpdatesToWsServer(websocketServer, builder)
val allMapTilesTable: KTable<TileUpdateRecordKey, MapTile> = allMapTilesTable(builder)
val tilePrototypesTable: KTable<PrototypeName, MapTilePrototype> = prototypesTable(builder)
val tilePrototypeColourTable: KTable<PrototypeName, Colour> = tilePrototypeColourTable(builder)
val webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> =
aggregateWebMapTiles(allMapTilesTable, tilePrototypesTable)
aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable)
saveTileImages(webMapTiles)

View file

@ -7,44 +7,44 @@ import org.apache.kafka.common.serialization.Deserializer
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serializer
inline fun <reified T> BinaryFormat.kafkaSerializer() =
Serializer { topic: String, data: T ->
runCatching {
encodeToByteArray(data)
}.getOrElse { e ->
println(
"""
Exception on encodeToString
Topic: $topic
topicData: $data
topicData as T: $data
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> BinaryFormat.kafkaDeserializer() =
Deserializer { topic: String, data: ByteArray ->
runCatching {
decodeFromByteArray<T>(data)
}.getOrElse { e ->
println(
"""
Exception on decodeFromString,
Topic: $topic
topicData: $data
topicData as T: ${data as? T}
""".trimIndent()
)
e.printStackTrace()
throw e
}
}
inline fun <reified T> BinaryFormat.serde() = object : Serde<T> {
override fun serializer(): Serializer<T> = kafkaSerializer()
override fun deserializer(): Deserializer<T> = kafkaDeserializer()
}
//
//inline fun <reified T> BinaryFormat.kafkaSerializer() =
// Serializer { topic: String, data: T ->
// runCatching {
// encodeToByteArray(data)
// }.getOrElse { e ->
// println(
// """
// Exception on encodeToString
// Topic: $topic
// topicData: $data
// topicData as T: $data
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
//inline fun <reified T> BinaryFormat.kafkaDeserializer() =
// Deserializer { topic: String, data: ByteArray ->
// runCatching {
// decodeFromByteArray<T>(data)
// }.getOrElse { e ->
// println(
// """
// Exception on decodeFromString,
// Topic: $topic
// topicData: $data
// topicData as T: ${data as? T}
// """.trimIndent()
// )
// e.printStackTrace()
// throw e
// }
// }
//
//inline fun <reified T> BinaryFormat.serde() = object : Serde<T> {
// override fun serializer(): Serializer<T> = kafkaSerializer()
// override fun deserializer(): Deserializer<T> = kafkaDeserializer()
//}

View file

@ -9,7 +9,6 @@ import dev.adamko.kafkatorio.events.schema.MapTilePosition
import dev.adamko.kafkatorio.events.schema.MapTiles
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.processor.serdes.serde
import dev.adamko.kotka.extensions.consumedAs
import dev.adamko.kotka.extensions.materializedWith
import dev.adamko.kotka.extensions.streams.filter
@ -75,12 +74,5 @@ fun allMapTilesTable(
"all-tiles-keyed-on-position",
materializedWith(kxsBinary.serde(), kxsBinary.serde())
)
.suppress(
Suppressed.untilTimeLimit(
Duration.ofSeconds(30),
Suppressed.BufferConfig
.maxRecords(30)
.emitEarlyWhenFull()
)
)
}

View file

@ -1,5 +1,8 @@
package dev.adamko.kafkatorio.processor.topology
import com.sksamuel.scrimage.color.RGBColor
import dev.adamko.kafkatorio.events.schema.Colour
import kotlin.math.roundToInt
import kotlinx.serialization.Serializable
@Serializable
@ -15,3 +18,12 @@ value class SurfaceIndex(
value class FactorioServerId(val id: String) {
override fun toString() = id
}
fun Colour.toRgbColor(): RGBColor {
return RGBColor(
red.roundToInt(),
green.roundToInt(),
blue.roundToInt(),
alpha.roundToInt(),
)
}

View file

@ -19,7 +19,7 @@ fun playerUpdatesToWsServer(
consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde())
)
.foreach { _, value ->
println("sending ${value.packetType} packet to websocket")
// println("sending ${value.packetType} packet to websocket")
websocketServer.sendMessage(jsonMapper.encodeToString(value))
}
}

View file

@ -1,11 +1,12 @@
package dev.adamko.kafkatorio.processor.topology
import com.sksamuel.scrimage.color.RGBColor
import dev.adamko.kafkatorio.events.schema.Colour
import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.processor.serdes.serde
import dev.adamko.kotka.extensions.consumedAs
import dev.adamko.kotka.extensions.materializedWith
import dev.adamko.kotka.extensions.streams.flatMap
@ -23,7 +24,31 @@ value class PrototypeName(val name: String) {
}
fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTilePrototype> {
//fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTilePrototype> {
// return builder.stream<FactorioPacketKey, FactorioPrototypes>(
// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
// consumedAs(
// "consume-all-prototypes-packets",
// jsonMapper.serde(),
// jsonMapper.serde()
// )
// )
// .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
// prototypes.prototypes
// .filterIsInstance<MapTilePrototype>()
// .map { PrototypeName(it.name) to it }
// }
// .toTable(
// materializedWith(
// kxsBinary.serde(),
// kxsBinary.serde()
// )
// )
//}
/** Get the [Colour] of each [MapTilePrototype] */
fun tilePrototypeColourTable(builder: StreamsBuilder): KTable<PrototypeName, Colour> {
return builder.stream<FactorioPacketKey, FactorioPrototypes>(
"kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all",
consumedAs(
@ -35,7 +60,9 @@ fun prototypesTable(builder: StreamsBuilder): KTable<PrototypeName, MapTileProto
.flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes ->
prototypes.prototypes
.filterIsInstance<MapTilePrototype>()
.map { PrototypeName(it.name) to it }
.map {
PrototypeName(it.name) to it.mapColour
}
}
.toTable(
materializedWith(

View file

@ -1,28 +1,37 @@
package dev.adamko.kafkatorio.processor.topology
import com.sksamuel.scrimage.color.RGBColor
import dev.adamko.kafkatorio.events.schema.Colour
import dev.adamko.kafkatorio.events.schema.MapTile
import dev.adamko.kafkatorio.events.schema.MapTilePosition
import dev.adamko.kafkatorio.events.schema.MapTilePrototype
import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition
import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kafkatorio.processor.serdes.serde
import dev.adamko.kotka.extensions.groupedAs
import dev.adamko.kotka.extensions.materializedAs
import dev.adamko.kotka.extensions.materializedWith
import dev.adamko.kotka.extensions.namedAs
import dev.adamko.kotka.extensions.streams.groupBy
import dev.adamko.kotka.extensions.tables.groupBy
import dev.adamko.kotka.extensions.tables.join
import dev.adamko.kotka.extensions.tables.mapValues
import dev.adamko.kotka.extensions.tables.toStream
import dev.adamko.kotka.extensions.toKeyValue
import dev.adamko.kotka.kxs.serde
import java.awt.Color
import java.time.Duration
import kotlinx.serialization.Serializable
import org.apache.kafka.streams.kstream.KTable
import org.apache.kafka.streams.kstream.SessionWindows
import org.apache.kafka.streams.kstream.Suppressed
import org.apache.kafka.streams.kstream.Windowed
/** The tiles of the webmap will be this many pixels high/wide */
const val WEB_MAP_IMAGE_TILE_SIZE = 256
/** Debounce duration */
private val WINDOW_INACTIVITY_DURATION = Duration.ofSeconds(60)
@Serializable
data class WebMapTileChunkPosition(
val chunkSize: Int,
@ -32,31 +41,52 @@ data class WebMapTileChunkPosition(
)
@Serializable
data class WebMapTilePixel(
private data class WebMapTilePixel(
val mapColour: Colour,
val tilePosition: MapTilePosition,
)
@Serializable
data class WebMapTileChunkPixels(
val pixels: Set<WebMapTilePixel> = mutableSetOf(),
@JvmInline
value class WebMapTileChunkPixels(
val pixels: Map<MapTilePosition, Colour>,
)
/** Temporary mutable aggregator */
/** Temporary mutable accumulator */
@Serializable
private data class WebMapTileChunkPixelsAcc(
val pixels: MutableSet<WebMapTilePixel> = mutableSetOf(),
)
@JvmInline
private value class WebMapTileChunkPixelsAcc(
val pixels: MutableMap<MapTilePosition, Colour> = mutableMapOf(),
) {
operator fun plus(pixel: WebMapTilePixel): WebMapTileChunkPixelsAcc {
pixels[pixel.tilePosition] = pixel.mapColour
return this
}
operator fun plus(other: WebMapTileChunkPixelsAcc): WebMapTileChunkPixelsAcc {
this.pixels += other.pixels
return this
}
operator fun minus(other: WebMapTileChunkPixelsAcc): WebMapTileChunkPixelsAcc {
this.pixels -= other.pixels.keys
return this
}
}
fun aggregateWebMapTiles(
allMapTilesTable: KTable<TileUpdateRecordKey, MapTile>,
tilePrototypesTable: KTable<PrototypeName, MapTilePrototype>,
tilePrototypeColourTable: KTable<PrototypeName, Colour>,
): KTable<WebMapTileChunkPosition, WebMapTileChunkPixels> {
val webmapTileColours: KTable<TileUpdateRecordKey, WebMapTilePixel> =
val pixels: KTable<TileUpdateRecordKey, WebMapTilePixel> =
allMapTilesTable
.join(
tilePrototypesTable,
tilePrototypeColourTable,
"convert-tiles-to-colours",
materializedAs(
"store-web-map-tile-pixels",
@ -64,13 +94,14 @@ fun aggregateWebMapTiles(
kxsBinary.serde(),
),
{ PrototypeName(it.prototypeName) },
) { tile: MapTile, proto: MapTilePrototype ->
WebMapTilePixel(proto.mapColour, tile.position)
) { tile: MapTile, colour: Colour ->
WebMapTilePixel(colour, tile.position)
}
val webMapTileChunkAggregate: KTable<WebMapTileChunkPosition, WebMapTileChunkPixelsAcc> =
webmapTileColours
val chunkedPixelsWindowed: KTable<Windowed<WebMapTileChunkPosition>, WebMapTileChunkPixelsAcc> =
pixels
.toStream("stream-webmap-tile-pixels")
.groupBy(
groupedAs(
"group-webmap-tile-colours-into-chunks",
@ -85,17 +116,21 @@ fun aggregateWebMapTiles(
chunkY,
pos.surfaceIndex
)
(chunkPos to px).toKeyValue()
chunkPos
}
.windowedBy(
SessionWindows.ofInactivityGapAndGrace(
WINDOW_INACTIVITY_DURATION,
WINDOW_INACTIVITY_DURATION,
)
)
.aggregate(
{ WebMapTileChunkPixelsAcc() },
/* adder */
{ _: WebMapTileChunkPosition, newValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc ->
acc.apply { pixels.add(newValue) }
acc + newValue
},
/* subtractor */
{ _: WebMapTileChunkPosition, oldValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc ->
acc.apply { pixels.minus(oldValue) }
{ _, aggOne: WebMapTileChunkPixelsAcc, aggTwo: WebMapTileChunkPixelsAcc ->
aggOne + aggTwo
},
namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"),
materializedAs(
@ -105,20 +140,34 @@ fun aggregateWebMapTiles(
)
)
.suppress(
Suppressed.untilTimeLimit<WebMapTileChunkPosition>(
Duration.ofSeconds(30),
Suppressed.BufferConfig
.maxRecords(30)
// .withMaxBytes(31457280 ) // 30MB
.emitEarlyWhenFull()
).withName("web-map-tile-aggregate-debounce")
Suppressed.untilWindowCloses(
Suppressed.BufferConfig.unbounded()
).withName("chunked-pixels-debounce")
)
return webMapTileChunkAggregate
// merge windows
val chunkedPixels: KTable<WebMapTileChunkPosition, WebMapTileChunkPixelsAcc> =
chunkedPixelsWindowed
.groupBy(
groupedAs("merge-chunked-pixels.group-keys", kxsBinary.serde(), kxsBinary.serde())
) { a: Windowed<WebMapTileChunkPosition>, b: WebMapTileChunkPixelsAcc ->
(a.key() to b).toKeyValue()
}
.reduce(
/* adder */
{ pixels1, pixels2 -> pixels1 + pixels2 },
/* subtractor */
{ pixels1, pixels2 -> pixels1 - pixels2 },
namedAs("merge-chunked-pixels.reduce-values"),
materializedAs("merge-chunked-pixels.reduce-values.store", kxsBinary.serde(), kxsBinary.serde())
)
return chunkedPixels
.mapValues(
"finalise-web-map-tile-colour-chunk-aggregation",
materializedAs("web-map-tile-colour-chunks", kxsBinary.serde(), kxsBinary.serde())
) { _, v ->
) { _: WebMapTileChunkPosition, v: WebMapTileChunkPixelsAcc ->
WebMapTileChunkPixels(v.pixels)
}
}

View file

@ -3,28 +3,24 @@ package dev.adamko.kafkatorio.processor.topology
import com.sksamuel.scrimage.ImmutableImage
import com.sksamuel.scrimage.color.RGBColor
import com.sksamuel.scrimage.nio.PngWriter
import dev.adamko.kafkatorio.events.schema.Colour
import dev.adamko.kafkatorio.events.schema.MapTilePosition
import dev.adamko.kotka.extensions.tables.toStream
import java.awt.Color
import java.awt.image.BufferedImage
import java.io.File
import java.time.Duration
import kotlin.math.abs
import kotlin.math.roundToInt
import org.apache.kafka.streams.kstream.KTable
import org.apache.kafka.streams.kstream.Suppressed
private val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f))
fun saveTileImages(
webMapTiles: KTable<WebMapTileChunkPosition, WebMapTileChunkPixels>
) {
webMapTiles
.suppress(
Suppressed.untilTimeLimit<WebMapTileChunkPosition>(
Duration.ofSeconds(30),
Suppressed.BufferConfig.maxRecords(30)
).withName("save-tile-images-debounce")
)
.toStream("get-webmap-tile-updates")
.foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels ->
runCatching {
@ -42,7 +38,7 @@ fun saveTileImages(
private fun saveMapTilesPng(
chunkPosition: WebMapTileChunkPosition,
pixels: Set<WebMapTilePixel>
pixels: Map<MapTilePosition, Colour>,
) {
val chunkOriginX: Int = chunkPosition.x * chunkPosition.chunkSize // - 1
val chunkOriginY: Int = chunkPosition.y * chunkPosition.chunkSize // - 1
@ -52,21 +48,21 @@ private fun saveMapTilesPng(
ImmutableImage.filled(
chunkPosition.chunkSize,
chunkPosition.chunkSize,
Color.BLACK,
COLOUR_TRANSPARENT.awt(),
BufferedImage.TYPE_INT_ARGB
)
pixels.forEach { pixel ->
pixels.forEach { (tilePosition, colour) ->
val rgbColour = RGBColor(
pixel.mapColour.red.roundToInt(),
pixel.mapColour.green.roundToInt(),
pixel.mapColour.blue.roundToInt(),
pixel.mapColour.alpha.roundToInt(),
colour.red.roundToInt(),
colour.green.roundToInt(),
colour.blue.roundToInt(),
colour.alpha.roundToInt(),
)
val pixelX = abs(abs(pixel.tilePosition.x) - abs(chunkOriginX))
val pixelY = abs(abs(pixel.tilePosition.y) - abs(chunkOriginY))
val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX))
val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY))
chunkImage.setColor(
pixelX,