some fixes

Author: Pawel Czajka

engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/ExtendedWindowOperator.scala

@@ -20,7 +20,7 @@ import pl.touk.nussknacker.engine.flink.api.process.FlinkCustomNodeContext
 import pl.touk.nussknacker.engine.flink.util.keyed.{KeyEnricher, StringKeyedValue}
 import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.ExtendedWindowOperator.{stateDescriptorName, Input}
 import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.transformers.AggregatorTypeInformations
-import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.triggers.{ClosingEndEventTrigger, FireOnEachEvent}
+import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.triggers.FireOnEachEvent
 
 import java.lang
 import java.util.concurrent.atomic.AtomicReference
@@ -31,17 +31,6 @@ object ExtendedWindowOperator {
   // WindowOperatorBuilder.WINDOW_STATE_NAME - should be the same for compatibility
   val stateDescriptorName = "window-contents"
 
-  def fireOnEachEventTriggerWrapper[T, W <: Window](delegate: Trigger[_ >: T, W]): FireOnEachEvent[T, W] = {
-    FireOnEachEvent(delegate)
-  }
-
-  def closingEndEventTriggerWrapper[T, W <: Window](
-      delegate: Trigger[_ >: T, W],
-      endFunction: T => Boolean
-  ): ClosingEndEventTrigger[T, W] = {
-    ClosingEndEventTrigger(delegate, endFunction)
-  }
-
   implicit class OnEventOperatorKeyedStream[A](stream: KeyedStream[Input[A], String])(
       implicit fctx: FlinkCustomNodeContext
   ) {
@@ -55,7 +44,7 @@ object ExtendedWindowOperator {
       assigner,
       types,
       aggregateFunction,
-      fireOnEachEventTriggerWrapper[ValueWithContext[StringKeyedValue[A]], TimeWindow](trigger),
+      FireOnEachEvent(trigger),
       preserveContext = true
     )
 
@@ -145,24 +134,25 @@ private class ValueEmittingWindowFunction(
       out: Collector[ValueWithContext[AnyRef]]
   ): Unit = {
     elements.forEach { element =>
-      contextHolderRef.get match {
+      val contextOpt = contextHolderRef.get match {
         case OnElementWindowContext(contextToPreserve, timestampToOverride) =>
+          // in current flink implementation out is always of type TimestampedCollector
           out.asInstanceOf[TimestampedCollector[_]].setAbsoluteTimestamp(timestampToOverride)
-          out.collect(
-            ValueWithContext(
-              element,
-              KeyEnricher.enrichWithKey(
-                contextToPreserve.getOrElse(api.Context(contextIdGenerator.nextContextId())),
-                key
-              )
-            )
-          )
+          contextToPreserve
 
         case OnTimerWindowContext =>
-          out.collect(
-            ValueWithContext(element, KeyEnricher.enrichWithKey(api.Context(contextIdGenerator.nextContextId()), key))
-          )
+          None
       }
+
+      out.collect(
+        ValueWithContext(
+          element,
+          KeyEnricher.enrichWithKey(
+            contextOpt.getOrElse(api.Context(contextIdGenerator.nextContextId())),
+            key
+          )
+        )
+      )
     }
   }
 

engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/transformers.scala

@@ -15,6 +15,7 @@ import pl.touk.nussknacker.engine.flink.api.process._
 import pl.touk.nussknacker.engine.flink.api.typeinformation.TypeInformationDetection
 import pl.touk.nussknacker.engine.flink.util.richflink._
 import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.ExtendedWindowOperator.OnEventOperatorKeyedStream
+import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.triggers.ClosingEndEventTrigger
 import pl.touk.nussknacker.engine.util.KeyedValue
 
 import scala.collection.immutable.SortedMap
@@ -176,11 +177,11 @@ object transformers {
           implicit val fctx: FlinkCustomNodeContext = ctx
           val typeInfos                             = AggregatorTypeInformations(ctx, aggregator, aggregateBy)
 
-          val baseTrigger = {
-            ExtendedWindowOperator.closingEndEventTriggerWrapper[ValueWithContext[KeyedValue[String, (AnyRef, java.lang.Boolean)]], TimeWindow](
+          val baseTrigger =
+            ClosingEndEventTrigger[ValueWithContext[KeyedValue[String, (AnyRef, java.lang.Boolean)]], TimeWindow](
               EventTimeTrigger.create(),
-              _.value.value._2)
-          }
+              _.value.value._2
+            )
           val groupByValue = aggregateBy.product(endSessionCondition)
 
           val keyedStream = start
@@ -193,7 +194,8 @@ object transformers {
             case SessionWindowTrigger.OnEvent =>
               keyedStream.extendedEventTriggerWindow(windowDefinition, typeInfos, aggregatingFunction, baseTrigger)
             case SessionWindowTrigger.OnEnd =>
-              keyedStream.extendedWindow(windowDefinition, typeInfos, aggregatingFunction, baseTrigger, preserveContext = false)
+              keyedStream
+                .extendedWindow(windowDefinition, typeInfos, aggregatingFunction, baseTrigger, preserveContext = false)
           }).setUidWithName(ctx, ExplicitUidInOperatorsSupport.defaultExplicitUidInStatefulOperators)
         })
       )

engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/triggers.scala

@@ -30,13 +30,12 @@ object triggers {
   case class FireOnEachEvent[T, W <: Window](delegate: Trigger[_ >: T, W]) extends DelegatingTrigger[T, W](delegate) {
 
     override def onElement(element: T, timestamp: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {
-      val previousResult = super.onElement(element, timestamp, window, ctx)
-      val result = previousResult match {
+      val result = super.onElement(element, timestamp, window, ctx)
+      result match {
         case TriggerResult.CONTINUE => TriggerResult.FIRE
         case TriggerResult.PURGE    => TriggerResult.FIRE_AND_PURGE
         case fire                   => fire
       }
-      result
     }
 
     override def onProcessingTime(time: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {
@@ -51,10 +50,8 @@ object triggers {
 
   }
 
-  case class ClosingEndEventTrigger[T, W <: Window](
-      delegate: Trigger[_ >: T, W],
-      endFunction: T => Boolean
-  ) extends DelegatingTrigger[T, W](delegate) {
+  case class ClosingEndEventTrigger[T, W <: Window](delegate: Trigger[_ >: T, W], endFunction: T => Boolean)
+      extends DelegatingTrigger[T, W](delegate) {
 
     override def onElement(element: T, timestamp: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {
       if (endFunction(element)) {