simplify DebounceProcessor - use for-loop and schedule instead of coroutines

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt

@@ -1,18 +1,17 @@
 package dev.adamko.kafkatorio.processor.misc
 
+import dev.adamko.kotka.extensions.component1
+import dev.adamko.kotka.extensions.component2
+import dev.adamko.kotka.extensions.component3
+import dev.adamko.kotka.extensions.processor.component1
+import dev.adamko.kotka.extensions.processor.component2
+import dev.adamko.kotka.extensions.processor.component3
 import dev.adamko.kotka.extensions.state.useAll
-import kotlin.coroutines.CoroutineContext
+import kotlin.time.toJavaDuration
-import kotlin.time.Duration
-import kotlinx.coroutines.CoroutineName
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.isActive
-import kotlinx.coroutines.launch
 import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.processor.Cancellable
+import org.apache.kafka.streams.processor.PunctuationType
 import org.apache.kafka.streams.processor.api.Processor
 import org.apache.kafka.streams.processor.api.ProcessorContext
 import org.apache.kafka.streams.processor.api.ProcessorSupplier
@@ -21,105 +20,116 @@ import org.apache.kafka.streams.state.StoreBuilder
 import org.apache.kafka.streams.state.Stores
 import org.apache.kafka.streams.state.TimestampedKeyValueStore
 import org.apache.kafka.streams.state.ValueAndTimestamp
+import java.time.Duration as DurationJvm
+import kotlin.time.Duration as DurationKt
 
 
 class DebounceProcessor<K, V>(
-  private val inactivityDuration: Duration,
+  private val inactivityDuration: DurationKt,
-  private val recordsTimestampedStoreName: String,
+  checkInterval: DurationKt,
+  private val storeName: String,
 ) : Processor<K, V, K, V> {
 
-  private var state: State<K, V> = Uninitialised()
+  private val checkInterval: DurationJvm = checkInterval.toJavaDuration()
 
-  override fun init(kafkaContext: ProcessorContext<K, V>) {
+  private lateinit var state: State<K, V>
-    state = Active(kafkaContext)
+
+  override fun init(context: ProcessorContext<K, V>) {
+    state = Active(context)
   }
 
   override fun process(record: Record<K, V>) {
-    when (val state = state) {
+    (state as Active).apply {
-      is Active -> state.addRecord(record)
+      val (key, value, _) = record
-      else      -> error("state is not Active (actual: $state)")
+
+      val existing: ValueAndTimestamp<V>? = store.get(key)
+      val newTimestamp = existing?.timestamp() ?: System.currentTimeMillis()
+
+      store.put(key, ValueAndTimestamp.make(value, newTimestamp))
     }
   }
 
   override fun close() {
-    (state as? Active)?.close()
+    (state as? Active)?.schedule?.cancel()
+    state = Closed()
   }
 
   private sealed interface State<K, V>
 
-  private class Uninitialised<K, V> : State<K, V>
-
   private class Closed<K, V> : State<K, V>
 
   private inner class Active(
-    val kafkaContext: ProcessorContext<K, V>,
+    val context: ProcessorContext<K, V>
-  ) : State<K, V>, CoroutineScope {
+  ) : State<K, V> {
-    override val coroutineContext: CoroutineContext =
+    val store: TimestampedKeyValueStore<K, V> = context.getStateStore(storeName)
-      SupervisorJob() +
+    val schedule: Cancellable =
-          Dispatchers.Unconfined +
+      context.schedule(checkInterval, PunctuationType.WALL_CLOCK_TIME, ::checkRecords)
-          CoroutineName("DebounceProcessor")
 
-    val recordsTimestampedStore: TimestampedKeyValueStore<K, V> =
+    private fun checkRecords(now: Long) {
-      kafkaContext.getStateStore(recordsTimestampedStoreName)
+      store.useAll {
-
+        forEach { record ->
-    private val recordJobs = mutableMapOf<K, TimestampedJob>()
+          val (key, value, timestamp) = record
-
+          if ((now - timestamp) > inactivityDuration.inWholeMilliseconds) {
-    init {
+            context.forward(Record(key, value, timestamp))
-      recordsTimestampedStore.useAll {
+            store.delete(key)
-        forEach { r ->
+          }
-          val record = Record(r.key, r.value.value(), r.value.timestamp())
-          addRecord(record)
         }
       }
     }
-
-    fun addRecord(record: Record<K, V>) {
-      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<K, V>): 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<K, V> {
-      coroutineContext.cancel()
-      return Closed()
-    }
   }
 
-  private data class TimestampedJob(
-    val job: Job,
-    val timestamp: Long,
-  )
-
-
   companion object {
+
+    fun <K, V> Topology.addDebounceProcessor(
+      namePrefix: String,
+      sourceTopic: String,
+      sinkTopic: String,
+
+      inactivityDuration: DurationKt,
+      checkInterval: DurationKt = inactivityDuration / 10,
+      keySerde: Serde<K>,
+      valueSerde: Serde<V>,
+      storeName: String = "$namePrefix.debounce-store",
+    ): Topology {
+
+      val sourceName = "$namePrefix.debounce-source"
+      val processorName = "$namePrefix.debounce-processor"
+      val sinkName = "$namePrefix.debounce-sink"
+
+      return addSource(
+        sourceName,
+        keySerde.deserializer(),
+        valueSerde.deserializer(),
+        sourceTopic,
+      ).addProcessor(
+        processorName,
+        Supplier(
+          inactivityDuration = inactivityDuration,
+          checkInterval = checkInterval,
+          storeName = storeName,
+          keySerde = keySerde,
+          valueSerde = valueSerde,
+        ),
+        sourceName,
+      ).addSink(
+        sinkName,
+        sinkTopic,
+        keySerde.serializer(),
+        valueSerde.serializer(),
+        processorName,
+      )
+    }
+
     class Supplier<K, V>(
-      private val inactivityDuration: Duration,
+      private val inactivityDuration: DurationKt,
+      private val checkInterval: DurationKt,
       private val storeName: String,
       private val keySerde: Serde<K>,
       private val valueSerde: Serde<V>,
     ) : ProcessorSupplier<K, V, K, V> {
 
-      override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
+      override fun get(): Processor<K, V, K, V> =
+        DebounceProcessor(inactivityDuration, checkInterval, storeName)
 
       override fun stores(): Set<StoreBuilder<*>> = setOf(
         Stores.timestampedKeyValueStoreBuilder(
@@ -131,181 +141,3 @@ class DebounceProcessor<K, V>(
     }
   }
 }
-
-
-//class DebounceProcessor<K, V>(
-//  private val inactivityDuration: Duration,
-//  private val recordsTimestampedStoreName: String,
-//  coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default
-//) :
-//  Processor<K, V, K, V>,
-//  CoroutineScope {
-//
-//  override var coroutineContext: CoroutineContext =
-//    SupervisorJob() +
-//        coroutineDispatcher +
-//        CoroutineName("DebounceProcessor")
-//
-//  private lateinit var kafkaContext: ProcessorContext<K, V>
-//  private lateinit var recordsTimestampedStore: TimestampedKeyValueStore<K, V?>
-//
-//  private val recordsFlow: MutableSharedFlow<DebounceTaskInitializer<K>> =
-//    MutableSharedFlow(
-//      replay = 0,
-//      extraBufferCapacity = Int.MAX_VALUE,
-//      onBufferOverflow = BufferOverflow.DROP_OLDEST,
-//    )
-//
-//  override fun init(context: ProcessorContext<K, V>) {
-//    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<K, Job>()) { 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<K>(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<K, V?>) {
-//    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<K>(
-//  val key: K,
-//  val createTask: Boolean,
-//) {
-//  constructor(keyValue: KeyValue<K, *>) : this(keyValue.key, keyValue.value != null)
-//  constructor(record: Record<K, *>) : this(record.key(), record.value() != null)
-//}
-//
-//
-//fun <K, V> debounceProcessor(
-//  inactivityDuration: Duration,
-//  storeName: String,
-//  keySerde: Serde<K>,
-//  valueSerde: Serde<V>,
-//): ProcessorSupplier<K, V, K, V> =
-//  object : ProcessorSupplier<K, V, K, V> {
-//
-//    override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
-//
-//    override fun stores(): Set<StoreBuilder<*>> = setOf(
-//      Stores.timestampedKeyValueStoreBuilder(
-//        Stores.persistentTimestampedKeyValueStore(storeName),
-//        keySerde,
-//        valueSerde,
-//      )
-//    )
-//
-//  }
-
-
-//class SuppressProcessor<K, V>(
-//  private val storeName: String,
-//  private val tickDuration: Duration = 1.seconds,
-//  private val debounceDuration: Duration,
-//) : Processor<K, V, K, V> {
-//
-//  private lateinit var context: ProcessorContext<K, V>
-//
-//  private lateinit var stateStore: TimestampedKeyValueStore<K, V?>
-//
-//  override fun init(context: ProcessorContext<K, V>) {
-//
-//    this.context = context
-//
-//    stateStore = context.getStateStore(storeName) as TimestampedKeyValueStore<K, V?>
-//
-//    context.schedule(
-//      tickDuration.toJavaDuration(),
-//      PunctuationType.WALL_CLOCK_TIME,
-//      ::punctuate,
-//    )
-//  }
-//
-//  override fun process(record: Record<K, V?>) {
-//
-//    val (key, newVal) = record
-//
-//    val storedVal: V? = stateStore[key]?.value()
-//
-//    val cachedVal = ValueAndTimestamp.make(
-//      storedVal ?: newVal,
-//      System.currentTimeMillis()
-//    )
-//
-//    stateStore.put(key, cachedVal)
-//
-////    val newTimestamp = if (storedVal == null) System.currentTimeMillis() else storedTimestamp
-////    stateStore.put(key, ValueAndTimestamp.make(storedVal, newTimestamp))
-//  }
-//
-//  private fun punctuate(currentTimestamp: Long) {
-//
-//    stateStore.all().use { iterator ->
-//      iterator.forEach { record: KeyValue<K, ValueAndTimestamp<V?>> ->
-//
-//        val (k, v, storedTimestamp) = record
-//
-//        if (currentTimestamp - storedTimestamp > debounceDuration.inWholeMilliseconds) {
-//
-//          context.forward(Record(k, v, storedTimestamp))
-//
-//          stateStore.delete(k)
-//        }
-//
-//      }
-//    }
-//  }
-//
-//}
-//