From 48adf0009775beaa090e6bbb0d9ffb858ced1aa6 Mon Sep 17 00:00:00 2001
From: Piotr Przybylski <piotrprz@gmail.com>
Date: Mon, 7 Oct 2024 08:50:30 +0200
Subject: [PATCH] Remove MultiMap (#6988)

---
 docs/Changelog.md                             |  1 +
 docs/MigrationGuide.md                        |  4 +-
 .../flink/api/state/EvictableState.scala      | 14 ------
 .../engine/flink/api/state/MultiMap.scala     | 44 -------------------
 4 files changed, 4 insertions(+), 59 deletions(-)
 delete mode 100644 engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/MultiMap.scala

diff --git a/docs/Changelog.md b/docs/Changelog.md
index 6b15d808e68..7a79052c450 100644
--- a/docs/Changelog.md
+++ b/docs/Changelog.md
@@ -64,6 +64,7 @@
 * [#6952](https://github.com/TouK/nussknacker/pull/6952) Improvement: TypeInformation support for scala.Option
 * [#6840](https://github.com/TouK/nussknacker/pull/6840) Introduce canCastTo, castTo and castToOrNull extension methods in SpeL.
 * [#6974](https://github.com/TouK/nussknacker/pull/6974) Add SpeL suggestions for cast methods parameter.
+* [#6988](https://github.com/TouK/nussknacker/pull/6988) Remove unused API classes: `MultiMap`, `TimestampedEvictableStateFunction`
 
 ## 1.17
 
diff --git a/docs/MigrationGuide.md b/docs/MigrationGuide.md
index 0dcbf40e175..767a5f42964 100644
--- a/docs/MigrationGuide.md
+++ b/docs/MigrationGuide.md
@@ -9,11 +9,13 @@ To see the biggest differences please consult the [changelog](Changelog.md).
 * [#6695](https://github.com/TouK/nussknacker/pull/6695) `SingleTypingResult` API changes:
   * Added `typeHintsObjType` which is used as a type for a type hints, suggester and validation.
   * Renamed `objType` to `runtimeObjType` which indicates a current object in a runtime.
-
 * [#6766](https://github.com/TouK/nussknacker/pull/6766)
   * Process API changes:
      * Field `ScenarioWithDetails.labels` was added
      * Field `ScenarioWithDetails.tags` was removed (it had the same value as `labels` and was not used)
+* [#6988](https://github.com/TouK/nussknacker/pull/6988) Removed unused API classes: `MultiMap`, `TimestampedEvictableStateFunction`.
+  `MultiMap` was incorrectly handled by Flink's default Kryo serializer, so if you want to copy it to your code
+  you should write and register a proper serializer.
 
 ### REST API changes
 
diff --git a/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala b/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala
index d698f9e42a9..45d17502995 100644
--- a/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala
+++ b/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala
@@ -47,20 +47,6 @@ abstract class EvictableStateFunction[In, Out, StateType] extends KeyedProcessFu
 
 }
 
-abstract class TimestampedEvictableStateFunction[In, Out, StateType]
-    extends EvictableStateFunction[In, Out, MultiMap[Long, StateType]] {
-
-  override protected def moveEvictionTime(offset: Long, ctx: KeyedProcessFunction[String, In, Out]#Context): Unit = {
-    super.moveEvictionTime(offset, ctx)
-    state.update(stateValue.from(ctx.timestamp() - offset))
-  }
-
-  protected def stateValue: MultiMap[Long, StateType] = {
-    Option(state.value()).getOrElse(MultiMap[Long, StateType](Ordering.Long))
-  }
-
-}
-
 abstract class LatelyEvictableStateFunction[In, Out, StateType]
     extends KeyedProcessFunction[String, In, Out]
     with LatelyEvictableStateFunctionMixin[StateType] {
diff --git a/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/MultiMap.scala b/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/MultiMap.scala
deleted file mode 100644
index e76172bee2f..00000000000
--- a/engine/flink/components-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/MultiMap.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-package pl.touk.nussknacker.engine.flink.api.state
-
-import scala.collection.immutable.TreeMap
-import scala.collection.compat._
-
-object MultiMap {
-  def apply[K: Ordering, V]: MultiMap[K, V] = MultiMap(TreeMap())
-}
-
-case class MultiMap[K, V](map: TreeMap[K, List[V]]) {
-
-  def add(key: K, value: V): MultiMap[K, V] = {
-    val newElement = map.get(key) match {
-      case Some(list) => value :: list
-      case None       => List(value)
-    }
-    MultiMap(map + (key -> newElement))
-  }
-
-  def add(key: K, values: List[V]): MultiMap[K, V] = {
-    val newElement = map.get(key) match {
-      case Some(list) => values ::: list
-      case None       => values
-    }
-    MultiMap(map + (key -> newElement))
-  }
-
-  def remove(key: K, value: V): MultiMap[K, V] = {
-    map.get(key) match {
-      case Some(list) =>
-        // TODO: this is only ineffective operation, but in our case lists should be rather short
-        val withRemovedEl = list.filterNot(_ == value)
-        MultiMap(map + (key -> withRemovedEl))
-      case None =>
-        this
-    }
-
-  }
-
-  def from(minimalKey: K) = MultiMap(map.rangeFrom(minimalKey))
-
-  def until(minimalKey: K) = MultiMap(map.rangeUntil(minimalKey))
-
-}