Optimize Delay component

TouK · Mar 27, 2024 · e04ca8f · e04ca8f
1 parent 4872f8a
commit e04ca8f
Showing 1 changed file with 1 addition and 3 deletions.
diff --git a/...d/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/DelayTransformer.scala b/...d/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/DelayTransformer.scala
@@ -15,7 +15,6 @@ import pl.touk.nussknacker.engine.flink.util.keyed.StringKeyOnlyMapper
 
 import java.time.Duration
 import java.util
-import java.util.Collections
 import javax.annotation.Nullable
 
 object DelayTransformer extends DelayTransformer
@@ -81,15 +80,14 @@ class DelayFunction(nodeCtx: FlinkCustomNodeContext, delay: Duration)
     val fireTime = ctx.timestamp() + delay.toMillis
 
     val currentState = readStateValueOrInitial(fireTime)
-    currentState.add(0, value.context)
+    currentState.add(value.context)
     state.put(fireTime, currentState)
 
     ctx.timerService().registerEventTimeTimer(fireTime)
   }
 
   override def onTimer(timestamp: Long, funCtx: FlinkTimerCtx, out: Collector[ValueWithContext[AnyRef]]): Unit = {
     val currentState = readStateValueOrInitial(timestamp)
-    Collections.reverse(currentState)
     currentState.forEach(emitValue(out, _))
     state.remove(timestamp)
   }