From 041d570aed849f320a622c5adc69ec36ce42d1d6 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 24 Apr 2022 18:55:18 +0200 Subject: [PATCH] re-implement debounce processor... and then realise it's difficult to integrate with Kafka Streams DSL so don't use it --- .../processor/misc/DebounceProcessor.kt | 313 ++++++++++++------ .../processor/misc/DebounceProcessorTest.kt | 22 +- 2 files changed, 217 insertions(+), 118 deletions(-) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt index 8598c46..d4bf09b 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt @@ -1,29 +1,18 @@ package dev.adamko.kafkatorio.processor.misc -import dev.adamko.kotka.extensions.component1 -import dev.adamko.kotka.extensions.component2 -import dev.adamko.kotka.extensions.state.allAsSequence +import dev.adamko.kotka.extensions.state.useAll import kotlin.coroutines.CoroutineContext import kotlin.time.Duration -import kotlinx.coroutines.CancellationException -import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.cancelChildren -import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.cancel import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.asFlow -import kotlinx.coroutines.flow.emitAll -import kotlinx.coroutines.flow.launchIn -import kotlinx.coroutines.flow.runningFold import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import org.apache.kafka.common.serialization.Serde -import org.apache.kafka.streams.KeyValue import org.apache.kafka.streams.processor.api.Processor import org.apache.kafka.streams.processor.api.ProcessorContext import org.apache.kafka.streams.processor.api.ProcessorSupplier @@ -37,117 +26,227 @@ import org.apache.kafka.streams.state.ValueAndTimestamp class DebounceProcessor( private val inactivityDuration: Duration, private val recordsTimestampedStoreName: String, - coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default, -) : - Processor, - CoroutineScope { +) : Processor { - override var coroutineContext: CoroutineContext = - SupervisorJob() + - coroutineDispatcher + - CoroutineName("DebounceProcessor") + private var state: State = Uninitialised() - private lateinit var kafkaContext: ProcessorContext - private lateinit var recordsTimestampedStore: TimestampedKeyValueStore - - private val recordsFlow: MutableSharedFlow> = - MutableSharedFlow( - replay = 0, - extraBufferCapacity = Int.MAX_VALUE, - onBufferOverflow = BufferOverflow.DROP_OLDEST, - ) - - override fun init(context: ProcessorContext) { - this.kafkaContext = context - - this.coroutineContext += CoroutineName("DebounceProcessor-${context.applicationId()}-${context.taskId()}") - - recordsTimestampedStore = kafkaContext.getStateStore(recordsTimestampedStoreName) - - // start collecting records - create a debounce-timer for each record - recordsFlow - .runningFold(mapOf()) { acc, (k, createTask) -> - require(coroutineContext.isActive) - - // cancel any previously existing debounce task - acc[k]?.cancel() - - // add a new debounce task - when (createTask) { - true -> acc.plus(k to createDebounceTask(k)) - false -> acc.minus(k) - } - - }.launchIn(this) - - // re-initialise the flow from previously-stored records - launch { - recordsFlow.emitAll( - recordsTimestampedStore - .allAsSequence() - .map { keyValue -> - require(coroutineContext.isActive) - DebounceTaskInitializer(keyValue) - } - .asFlow() - ) - } + override fun init(kafkaContext: ProcessorContext) { + state = Active(kafkaContext) } - private fun createDebounceTask(key: K): Job { - return launch { - delay(inactivityDuration) - val (value, timestamp) = recordsTimestampedStore.get(key) - if (coroutineContext.isActive) { - kafkaContext.forward(Record(key, value, timestamp)) - } - } - } - - override fun process(record: Record) { - launch { - recordsTimestampedStore.put( - record.key(), - ValueAndTimestamp.make(record.value(), record.timestamp()) - ) - recordsFlow.emit(DebounceTaskInitializer(record)) + override fun process(record: Record) { + when (val state = state) { + is Active -> state.addRecord(record) + else -> error("state is not Active (actual: $state)") } } override fun close() { - coroutineContext.cancelChildren(CancellationException("processor go sleepy bye bye")) + (state as? Active)?.close() } -} + private sealed interface State -internal data class DebounceTaskInitializer( - val key: K, - val createTask: Boolean, -) { - constructor(keyValue: KeyValue) : this(keyValue.key, keyValue.value != null) - constructor(record: Record) : this(record.key(), record.value() != null) -} + private class Uninitialised : State + + private class Closed : State + + private inner class Active( + val kafkaContext: ProcessorContext, + ) : State, CoroutineScope { + override val coroutineContext: CoroutineContext = + SupervisorJob() + + Dispatchers.Unconfined + + CoroutineName("DebounceProcessor") + + val recordsTimestampedStore: TimestampedKeyValueStore = + kafkaContext.getStateStore(recordsTimestampedStoreName) + + private val recordJobs = mutableMapOf() + + init { + recordsTimestampedStore.useAll { + forEach { r -> + val record = Record(r.key, r.value.value(), r.value.timestamp()) + addRecord(record) + } + } + } + + fun addRecord(record: Record) { + recordJobs.compute(record.key()) { _, existing -> + if (existing != null && existing.timestamp < record.timestamp()) { + existing.job.cancel("replaced by newer record ${existing.timestamp} < ${record.timestamp()}") + createTimestampedJob(record) + } else { + null + } + } + } + + private fun createTimestampedJob(record: Record): TimestampedJob { + val job = launch { + recordsTimestampedStore.put( + record.key(), + ValueAndTimestamp.make(record.value(), record.timestamp()) + ) + delay(inactivityDuration) + if (isActive) { + kafkaContext.forward(record) + recordsTimestampedStore.delete(record.key()) + } + } + return TimestampedJob(job, record.timestamp()) + } + + fun close(): Closed { + coroutineContext.cancel() + return Closed() + } + } + + private data class TimestampedJob( + val job: Job, + val timestamp: Long, + ) -fun debounceProcessor( - inactivityDuration: Duration, - storeName: String, - keySerde: Serde, - valueSerde: Serde, -): ProcessorSupplier = - object : ProcessorSupplier { + companion object { + class Supplier( + private val inactivityDuration: Duration, + private val storeName: String, + private val keySerde: Serde, + private val valueSerde: Serde, + ) : ProcessorSupplier { - override fun get(): Processor = DebounceProcessor(inactivityDuration, storeName) + override fun get(): Processor = DebounceProcessor(inactivityDuration, storeName) - override fun stores(): Set> = setOf( - Stores.timestampedKeyValueStoreBuilder( - Stores.persistentTimestampedKeyValueStore(storeName), - keySerde, - valueSerde, + override fun stores(): Set> = setOf( + Stores.timestampedKeyValueStoreBuilder( + Stores.persistentTimestampedKeyValueStore(storeName), + keySerde, + valueSerde, + ) ) - ) - + } } +} + + +//class DebounceProcessor( +// private val inactivityDuration: Duration, +// private val recordsTimestampedStoreName: String, +// coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default +//) : +// Processor, +// CoroutineScope { +// +// override var coroutineContext: CoroutineContext = +// SupervisorJob() + +// coroutineDispatcher + +// CoroutineName("DebounceProcessor") +// +// private lateinit var kafkaContext: ProcessorContext +// private lateinit var recordsTimestampedStore: TimestampedKeyValueStore +// +// private val recordsFlow: MutableSharedFlow> = +// MutableSharedFlow( +// replay = 0, +// extraBufferCapacity = Int.MAX_VALUE, +// onBufferOverflow = BufferOverflow.DROP_OLDEST, +// ) +// +// override fun init(context: ProcessorContext) { +// this.kafkaContext = context +// +// this.coroutineContext += CoroutineName("DebounceProcessor-${context.applicationId()}-${context.taskId()}") +// +// recordsTimestampedStore = kafkaContext.getStateStore(recordsTimestampedStoreName) +// +// // start collecting records - create a debounce-timer for each record +// recordsFlow +// .runningFold(mapOf()) { acc, (k, createTask) -> +// require(coroutineContext.isActive) +// +// // cancel any previously existing debounce task +// acc[k]?.cancel() +// +// // add a new debounce task +// when (createTask) { +// true -> acc.plus(k to createDebounceTask(k)) +// false -> acc.minus(k) +// } +// +// }.launchIn(this) +// +// // re-initialise the flow from previously-stored records +// launch { +// recordsFlow.emitAll( +// recordsTimestampedStore +// .allAsSequence() +// .map { keyValue -> +// require(coroutineContext.isActive) +// DebounceTaskInitializer(keyValue) +// } +// .asFlow() +// ) +// } +// } +// +// private fun createDebounceTask(key: K): Job { +// return launch { +// delay(inactivityDuration) +// val (value, timestamp) = recordsTimestampedStore.get(key) +// if (coroutineContext.isActive) { +// kafkaContext.forward(Record(key, value, timestamp)) +// } +// } +// } +// +// override fun process(record: Record) { +// launch { +// recordsTimestampedStore.put( +// record.key(), +// ValueAndTimestamp.make(record.value(), record.timestamp()) +// ) +// recordsFlow.emit(DebounceTaskInitializer(record)) +// } +// } +// +// override fun close() { +// coroutineContext.cancelChildren(CancellationException("processor go sleepy bye bye")) +// } +// +//} +// +//internal data class DebounceTaskInitializer( +// val key: K, +// val createTask: Boolean, +//) { +// constructor(keyValue: KeyValue) : this(keyValue.key, keyValue.value != null) +// constructor(record: Record) : this(record.key(), record.value() != null) +//} +// +// +//fun debounceProcessor( +// inactivityDuration: Duration, +// storeName: String, +// keySerde: Serde, +// valueSerde: Serde, +//): ProcessorSupplier = +// object : ProcessorSupplier { +// +// override fun get(): Processor = DebounceProcessor(inactivityDuration, storeName) +// +// override fun stores(): Set> = setOf( +// Stores.timestampedKeyValueStoreBuilder( +// Stores.persistentTimestampedKeyValueStore(storeName), +// keySerde, +// valueSerde, +// ) +// ) +// +// } //class SuppressProcessor( diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt index b7ff622..a8d777d 100644 --- a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt +++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt @@ -19,17 +19,17 @@ class DebounceProcessorTest : FunSpec({ fun blah() { - val topology = Topology() - - topology.addProcessor( - "blah", - debounceProcessor( - 1.minutes, - "asdasda", - Serdes.String(), - Serdes.String() - ) - ) +// val topology = Topology() +// +// topology.addProcessor( +// "blah", +// debounceProcessor( +// 1.minutes, +// "asdasda", +// Serdes.String(), +// Serdes.String() +// ) +// ) // val streamsBuilder = StreamsBuilder()