[Nu-8656] Abstract FlinkEngineRuntimeContext

DeamonDev · DeamonDev · commit 66e1ea1cb797 · 2025-10-07T10:52:01.000+02:00
diff --git a/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/FlinkEngineContext.scala b/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/FlinkEngineContext.scala
@@ -0,0 +1,34 @@
+package pl.touk.nussknacker.engine.flink.api
+
+import org.apache.flink.api.common.TaskInfo
+import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.api.connector.sink2.WriterInitContext
+import org.apache.flink.metrics.groups.OperatorMetricGroup
+import pl.touk.nussknacker.engine.flink.api.exception.ExceptionHandler
+
+sealed trait FlinkEngineContext
+final case class RuntimeCtx(ctx: RuntimeContext)       extends FlinkEngineContext
+final case class WriterInitCtx(ctx: WriterInitContext) extends FlinkEngineContext
+
+object FlinkEngineContextOps {
+
+  implicit class RichExceptionHandler(val f: FlinkEngineContext => ExceptionHandler) extends AnyVal {
+    def narrowToRuntimeCtx: RuntimeContext => ExceptionHandler =
+      rc => f(RuntimeCtx(rc))
+  }
+
+  implicit class RichFlinkEngineContext(val ctx: FlinkEngineContext) extends AnyVal {
+
+    def getTaskInfo: TaskInfo = ctx match {
+      case RuntimeCtx(ctx)    => ctx.getTaskInfo
+      case WriterInitCtx(ctx) => ctx.getTaskInfo
+    }
+
+    def getMetricGroup: OperatorMetricGroup = ctx match {
+      case RuntimeCtx(ctx)    => ctx.getMetricGroup
+      case WriterInitCtx(ctx) => ctx.metricGroup()
+    }
+
+  }
+
+}
diff --git a/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/FlinkEngineRuntimeContext.scala b/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/FlinkEngineRuntimeContext.scala
@@ -4,5 +4,5 @@ import org.apache.flink.api.common.functions.RuntimeContext
 import pl.touk.nussknacker.engine.api.runtimecontext.EngineRuntimeContext
 
 trait FlinkEngineRuntimeContext extends EngineRuntimeContext {
-  def runtimeContext: RuntimeContext
+  def runtimeContext: FlinkEngineContext
 }
diff --git a/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/exception/SinkWriterWithExceptionHandler.scala b/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/exception/SinkWriterWithExceptionHandler.scala
@@ -0,0 +1,17 @@
+package pl.touk.nussknacker.engine.flink.api.exception
+
+import org.apache.flink.api.connector.sink2.SinkWriter
+
+trait SinkWriterWithExceptionHandler[T] {
+  self: SinkWriter[T] =>
+
+  protected val exceptionHandler: ExceptionHandler
+
+  override def close(): Unit = {
+    self.close()
+    if (exceptionHandler != null) {
+      exceptionHandler.close()
+    }
+  }
+
+}
diff --git a/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/process/FlinkCustomNodeContext.scala b/engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/process/FlinkCustomNodeContext.scala
@@ -7,7 +7,7 @@ import pl.touk.nussknacker.engine.api.context.ValidationContext
 import pl.touk.nussknacker.engine.api.process.ComponentUseContext
 import pl.touk.nussknacker.engine.api.runtimecontext.EngineRuntimeContext
 import pl.touk.nussknacker.engine.api.typed.typing.{Typed, TypingResult, Unknown}
-import pl.touk.nussknacker.engine.flink.api.NkGlobalParameters
+import pl.touk.nussknacker.engine.flink.api.{FlinkEngineContext, NkGlobalParameters}
 import pl.touk.nussknacker.engine.flink.api.exception.ExceptionHandler
 import pl.touk.nussknacker.engine.flink.api.typeinformation.TypeInformationDetection
 
@@ -21,7 +21,7 @@ case class FlinkCustomNodeContext(
     timeout: FiniteDuration,
     convertToEngineRuntimeContext: RuntimeContext => EngineRuntimeContext,
     lazyParameterHelper: FlinkLazyParameterFunctionHelper,
-    exceptionHandlerPreparer: RuntimeContext => ExceptionHandler,
+    exceptionHandlerPreparer: FlinkEngineContext => ExceptionHandler,
     globalParameters: Option[NkGlobalParameters],
     validationContext: Either[ValidationContext, Map[String, ValidationContext]],
     componentUseContext: ComponentUseContext,
diff --git a/engine/flink/components/table/src/main/scala/pl/touk/nussknacker/engine/flink/table/sink/TableSink.scala b/engine/flink/components/table/src/main/scala/pl/touk/nussknacker/engine/flink/table/sink/TableSink.scala
@@ -12,6 +12,7 @@ import org.apache.flink.util.Collector
 import pl.touk.nussknacker.engine.api.{Context, LazyParameter, NodeId, ValueWithContext}
 import pl.touk.nussknacker.engine.api.component.{ComponentType, NodeComponentInfo}
 import pl.touk.nussknacker.engine.api.typed.typing.TypingResult
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContextOps._
 import pl.touk.nussknacker.engine.flink.api.exception.{ExceptionHandler, WithExceptionHandler}
 import pl.touk.nussknacker.engine.flink.api.process.{FlinkCustomNodeContext, FlinkSink}
 import pl.touk.nussknacker.engine.flink.api.typeinformation.TypeInformationDetection
@@ -105,7 +106,7 @@ object EncodeAsTableTypeFunction {
     val alignedType  = ToTableTypeSchemaBasedEncoder.alignTypingResult(valueReturnType, sinkRowType)
     val producedType = TypeInformationDetection.instance.forType[Row](alignedType)
     new EncodeAsTableTypeFunction(
-      flinkNodeContext.exceptionHandlerPreparer,
+      flinkNodeContext.exceptionHandlerPreparer.narrowToRuntimeCtx,
       flinkNodeContext.nodeId,
       sinkRowType,
       producedType
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/ProcessPartFunction.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/ProcessPartFunction.scala
@@ -1,6 +1,7 @@
 package pl.touk.nussknacker.engine.process
 
 import org.apache.flink.api.common.functions.{OpenContext, RichFunction}
+import pl.touk.nussknacker.engine.flink.api.RuntimeCtx
 import pl.touk.nussknacker.engine.graph.node.NodeData
 import pl.touk.nussknacker.engine.process.compiler.FlinkProcessCompilerData
 import pl.touk.nussknacker.engine.process.exception.FlinkExceptionHandler
@@ -45,7 +46,7 @@ trait ExceptionHandlerFunction extends RichFunction {
 
   override def open(openContext: OpenContext): Unit = {
     compilerData = compilerDataForClassloader(getRuntimeContext.getUserCodeClassLoader)
-    exceptionHandler = compilerData.prepareExceptionHandler(getRuntimeContext)
+    exceptionHandler = compilerData.prepareExceptionHandler(RuntimeCtx(getRuntimeContext))
   }
 
 }
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkEngineRuntimeContextImpl.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkEngineRuntimeContextImpl.scala
@@ -4,13 +4,14 @@ import org.apache.flink.api.common.functions.RuntimeContext
 import pl.touk.nussknacker.engine.RuntimeMode
 import pl.touk.nussknacker.engine.api.{JobData, NodeId}
 import pl.touk.nussknacker.engine.api.runtimecontext.{ContextIdGenerator, IncContextIdGenerator}
-import pl.touk.nussknacker.engine.flink.api.FlinkEngineRuntimeContext
+import pl.touk.nussknacker.engine.flink.api.{FlinkEngineContext, FlinkEngineRuntimeContext, RuntimeCtx}
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContextOps.RichFlinkEngineContext
 import pl.touk.nussknacker.engine.process.compiler.MetricsProviderForFlink.createMetricsProvider
 import pl.touk.nussknacker.engine.util.metrics.MetricsProviderForScenario
 
 case class FlinkEngineRuntimeContextImpl(
     jobData: JobData,
-    runtimeContext: RuntimeContext,
+    runtimeContext: FlinkEngineContext,
     metricsProvider: MetricsProviderForScenario
 ) extends FlinkEngineRuntimeContext {
 
@@ -28,7 +29,7 @@ object FlinkEngineRuntimeContextImpl {
 //  This creates FlinkEngineRuntimeContextImpl with correct metricsProviderForScenario based on ComponentUseContextProvider
   def apply(
       jobData: JobData,
-      runtimeContext: RuntimeContext,
+      runtimeContext: FlinkEngineContext,
       runtimeMode: RuntimeMode
   ): FlinkEngineRuntimeContextImpl = {
     val properMetricsProvider = createMetricsProvider(runtimeMode, runtimeContext)
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkMetricsProviderForScenario.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkMetricsProviderForScenario.scala
@@ -4,14 +4,16 @@ import cats.data.NonEmptyList
 import com.codahale.metrics
 import com.codahale.metrics.SlidingTimeWindowReservoir
 import org.apache.flink
-import org.apache.flink.api.common.functions.RuntimeContext
 import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper
 import org.apache.flink.metrics.MetricGroup
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContext
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContextOps.RichFlinkEngineContext
 import pl.touk.nussknacker.engine.util.metrics._
 
 import java.util.concurrent.TimeUnit
 
-class FlinkMetricsProviderForScenario(runtimeContext: RuntimeContext) extends BaseMetricsProviderForScenario {
+class FlinkMetricsProviderForScenario(exceptionHandlerContext: FlinkEngineContext)
+    extends BaseMetricsProviderForScenario {
 
   override def registerGauge[T](identifier: MetricIdentifier, value: Gauge[T]): Unit =
     gauge[T, flink.metrics.Gauge[T]](identifier.name, identifier.tags, () => value.getValue)
@@ -58,8 +60,9 @@ class FlinkMetricsProviderForScenario(runtimeContext: RuntimeContext) extends Ba
   private def tagMode(nameParts: NonEmptyList[String], tags: Map[String, String]): (MetricGroup, String) = {
     val lastName = nameParts.last
     // all but last
-    val metricNameParts    = nameParts.init
-    val groupWithNameParts = metricNameParts.foldLeft[MetricGroup](runtimeContext.getMetricGroup)(_.addGroup(_))
+    val metricNameParts = nameParts.init
+    val groupWithNameParts =
+      metricNameParts.foldLeft[MetricGroup](exceptionHandlerContext.getMetricGroup)(_.addGroup(_))
 
     val finalGroup = tags.toList.sortBy(_._1).foldLeft[MetricGroup](groupWithNameParts) {
       case (group, (tag, tagValue)) => group.addGroup(tag, tagValue)
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkProcessCompilerData.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkProcessCompilerData.scala
@@ -4,6 +4,7 @@ import cats.data._
 import cats.data.Validated.{Invalid, Valid}
 import org.apache.flink.api.common.functions.RuntimeContext
 import org.apache.flink.api.common.restartstrategy.RestartStrategies
+import org.apache.flink.api.connector.sink2.WriterInitContext
 import pl.touk.nussknacker.engine.{Interpreter, RuntimeMode, ScenarioCompilationDependencies}
 import pl.touk.nussknacker.engine.api.JobData
 import pl.touk.nussknacker.engine.api.context.{ProcessCompilationError, ScenarioCompilationErrors, ValidationContext}
@@ -18,6 +19,7 @@ import pl.touk.nussknacker.engine.compile.nodecompilation.{
 }
 import pl.touk.nussknacker.engine.compiledgraph.CompiledProcessParts
 import pl.touk.nussknacker.engine.compiledgraph.node.Node
+import pl.touk.nussknacker.engine.flink.api.{FlinkEngineContext, RuntimeCtx}
 import pl.touk.nussknacker.engine.graph.node.NodeData
 import pl.touk.nussknacker.engine.process.exception.FlinkExceptionHandler
 import pl.touk.nussknacker.engine.splittedgraph.splittednode.SplittedNode
@@ -43,7 +45,7 @@ class FlinkProcessCompilerData(
   def open(runtimeContext: RuntimeContext, nodesToUse: List[_ <: NodeData]): Unit = {
     val lifecycle = compilerData.lifecycle(nodesToUse)
     lifecycle.foreach {
-      _.open(FlinkEngineRuntimeContextImpl(jobData, runtimeContext, runtimeMode))
+      _.open(FlinkEngineRuntimeContextImpl(jobData, RuntimeCtx(runtimeContext), runtimeMode))
     }
   }
 
@@ -94,9 +96,9 @@ class FlinkProcessCompilerData(
 
   def restartStrategy: RestartStrategies.RestartStrategyConfiguration = exceptionHandler.restartStrategy
 
-  def prepareExceptionHandler(runtimeContext: RuntimeContext): FlinkExceptionHandler = {
+  def prepareExceptionHandler(flinkEngineContext: FlinkEngineContext): FlinkExceptionHandler = {
     exceptionHandler.open(
-      FlinkEngineRuntimeContextImpl(jobData, runtimeContext, runtimeMode)
+      FlinkEngineRuntimeContextImpl(jobData, flinkEngineContext, runtimeMode)
     )
     exceptionHandler
   }
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/MetricsProviderForFlink.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/MetricsProviderForFlink.scala
@@ -2,17 +2,18 @@ package pl.touk.nussknacker.engine.process.compiler
 
 import org.apache.flink.api.common.functions.RuntimeContext
 import pl.touk.nussknacker.engine.RuntimeMode
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContext
 import pl.touk.nussknacker.engine.util.metrics.{MetricsProviderForScenario, NoOpMetricsProviderForScenario}
 
 object MetricsProviderForFlink {
 
   def createMetricsProvider(
       runtimeMode: RuntimeMode,
-      runtimeContext: RuntimeContext
+      flinkEngineContext: FlinkEngineContext
   ): MetricsProviderForScenario = {
     runtimeMode match {
       case RuntimeMode.Test => NoOpMetricsProviderForScenario
-      case _                => new FlinkMetricsProviderForScenario(runtimeContext)
+      case _                => new FlinkMetricsProviderForScenario(flinkEngineContext)
     }
   }
 
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/registrar/FlinkProcessRegistrar.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/registrar/FlinkProcessRegistrar.scala
@@ -3,6 +3,7 @@ package pl.touk.nussknacker.engine.process.registrar
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.flink.api.common.functions.RuntimeContext
 import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.connector.sink2.WriterInitContext
 import org.apache.flink.streaming.api.datastream.{AsyncDataStream, DataStream, SingleOutputStreamOperator}
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
 import org.apache.flink.streaming.api.windowing.time.Time
@@ -16,7 +17,8 @@ import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.compiledgraph.part._
 import pl.touk.nussknacker.engine.deployment.DeploymentData
 import pl.touk.nussknacker.engine.flink.FlinkScenarioCompilationDependencies
-import pl.touk.nussknacker.engine.flink.api.NkGlobalParameters
+import pl.touk.nussknacker.engine.flink.api.{FlinkEngineContext, NkGlobalParameters, RuntimeCtx, WriterInitCtx}
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContextOps._
 import pl.touk.nussknacker.engine.flink.api.compat.ExplicitUidInOperatorsSupport
 import pl.touk.nussknacker.engine.flink.api.process._
 import pl.touk.nussknacker.engine.flink.api.typeinformation.TypeInformationDetection
@@ -124,19 +126,31 @@ class FlinkProcessRegistrar(
         nodeComponentId: NodeComponentInfo,
         validationContext: Either[ValidationContext, Map[String, ValidationContext]]
     ): FlinkCustomNodeContext = {
-      val exceptionHandlerPreparer = (runtimeContext: RuntimeContext) =>
-        compilerDataForProcessPart(None)(runtimeContext.getUserCodeClassLoader).prepareExceptionHandler(runtimeContext)
+      val runtimeCtxExceptionHandlerPreparer = (runtimeContext: RuntimeContext) =>
+        compilerDataForProcessPart(None)(runtimeContext.getUserCodeClassLoader).prepareExceptionHandler(
+          RuntimeCtx(runtimeContext)
+        )
+      val exceptionHandlerPreparer = (flinkEngineContext: FlinkEngineContext) =>
+        flinkEngineContext match {
+          case RuntimeCtx(runtimeContext) =>
+            runtimeCtxExceptionHandlerPreparer(runtimeContext)
+          case WriterInitCtx(writerInitContext) =>
+            compilerDataForProcessPart(None)(writerInitContext.getUserCodeClassLoader.asClassLoader())
+              .prepareExceptionHandler(WriterInitCtx(writerInitContext))
+
+        }
       val jobData                     = compilerData.jobData
       val componentUseContextProvider = compilerData.runtimeMode
 
       FlinkCustomNodeContext(
         jobData,
         nodeComponentId.nodeId,
         compilerData.processTimeout,
-        convertToEngineRuntimeContext = FlinkEngineRuntimeContextImpl(jobData, _, componentUseContextProvider),
+        convertToEngineRuntimeContext =
+          (r: RuntimeContext) => FlinkEngineRuntimeContextImpl(jobData, RuntimeCtx(r), componentUseContextProvider),
         lazyParameterHelper = new FlinkLazyParameterFunctionHelper(
           nodeComponentId,
-          exceptionHandlerPreparer,
+          runtimeCtxExceptionHandlerPreparer,
           createInterpreter(compilerDataForProcessPart(None))
         ),
         exceptionHandlerPreparer = exceptionHandlerPreparer,
diff --git a/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/registrar/SourceMetricsFunction.scala b/engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/registrar/SourceMetricsFunction.scala
@@ -4,6 +4,7 @@ import org.apache.flink.api.common.functions.OpenContext
 import org.apache.flink.streaming.api.functions.ProcessFunction
 import org.apache.flink.util.Collector
 import pl.touk.nussknacker.engine.RuntimeMode
+import pl.touk.nussknacker.engine.flink.api.RuntimeCtx
 import pl.touk.nussknacker.engine.process.compiler.MetricsProviderForFlink.createMetricsProvider
 import pl.touk.nussknacker.engine.util.metrics.common.OneSourceMetrics
 
@@ -16,7 +17,7 @@ private[registrar] class SourceMetricsFunction[T](
 
   override def open(openContext: OpenContext): Unit = {
     metrics = new OneSourceMetrics(sourceId)
-    val metricsProvider = createMetricsProvider(componentUseContextProvider, getRuntimeContext)
+    val metricsProvider = createMetricsProvider(componentUseContextProvider, RuntimeCtx(getRuntimeContext))
     metrics.registerOwnMetrics(metricsProvider)
   }
 
diff --git a/engine/flink/kafka-components-utils/src/main/scala/pl/touk/nussknacker/engine/kafka/source/flink/FlinkKafkaSource.scala b/engine/flink/kafka-components-utils/src/main/scala/pl/touk/nussknacker/engine/kafka/source/flink/FlinkKafkaSource.scala
@@ -22,6 +22,7 @@ import pl.touk.nussknacker.engine.api.process.{
 }
 import pl.touk.nussknacker.engine.api.runtimecontext.{ContextIdGenerator, EngineRuntimeContext}
 import pl.touk.nussknacker.engine.api.test.{TestData, TestRecord, TestRecordParser}
+import pl.touk.nussknacker.engine.flink.api.{FlinkEngineContext, RuntimeCtx}
 import pl.touk.nussknacker.engine.flink.api.compat.ExplicitUidInOperatorsSupport
 import pl.touk.nussknacker.engine.flink.api.exception.ExceptionHandler
 import pl.touk.nussknacker.engine.flink.api.process._
@@ -276,7 +277,7 @@ class FlinkKafkaConsumerHandlingExceptions[T](
     topics: java.util.List[String],
     deserializationSchema: FlinkDeserializationSchemaWrapper[T],
     props: Properties,
-    exceptionHandlerPreparer: RuntimeContext => ExceptionHandler,
+    exceptionHandlerPreparer: FlinkEngineContext => ExceptionHandler,
     convertToEngineRuntimeContext: RuntimeContext => EngineRuntimeContext,
     nodeId: NodeId
 ) extends FlinkKafkaConsumer[T](topics, deserializationSchema, props)
@@ -289,7 +290,7 @@ class FlinkKafkaConsumerHandlingExceptions[T](
   override def open(openContext: OpenContext): Unit = {
     patchRestoredState()
     super.open(openContext)
-    exceptionHandler = exceptionHandlerPreparer(getRuntimeContext)
+    exceptionHandler = exceptionHandlerPreparer(RuntimeCtx(getRuntimeContext))
     exceptionPurposeContextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId)
     deserializationSchema.setExceptionHandlingData(exceptionHandler, exceptionPurposeContextIdGenerator, nodeId)
   }
diff --git a/engine/flink/schemed-kafka-components-utils/src/main/scala/pl/touk/nussknacker/engine/schemedkafka/sink/flink/FlinkKafkaUniversalSink.scala b/engine/flink/schemed-kafka-components-utils/src/main/scala/pl/touk/nussknacker/engine/schemedkafka/sink/flink/FlinkKafkaUniversalSink.scala
@@ -11,6 +11,7 @@ import pl.touk.nussknacker.engine.api.{Context, LazyParameter, ValueWithContext}
 import pl.touk.nussknacker.engine.api.component.{ComponentType, NodeComponentInfo}
 import pl.touk.nussknacker.engine.api.process.TopicName
 import pl.touk.nussknacker.engine.api.validation.ValidationMode
+import pl.touk.nussknacker.engine.flink.api.FlinkEngineContextOps._
 import pl.touk.nussknacker.engine.flink.api.exception.{ExceptionHandler, WithExceptionHandler}
 import pl.touk.nussknacker.engine.flink.api.process.{FlinkCustomNodeContext, FlinkSink}
 import pl.touk.nussknacker.engine.flink.typeinformation.KeyedValueType
@@ -79,7 +80,7 @@ class FlinkKafkaUniversalSink(
     private val nodeId = flinkNodeContext.nodeId
 
     protected override val exceptionHandlerPreparer: RuntimeContext => ExceptionHandler =
-      flinkNodeContext.exceptionHandlerPreparer
+      flinkNodeContext.exceptionHandlerPreparer.narrowToRuntimeCtx
 
     @transient private var encodeRecord: Any => AnyRef = _
 

Original file line number	Diff line number	Diff line change
`@@ -4,5 +4,5 @@ import org.apache.flink.api.common.functions.RuntimeContext`
`4`	`4`	`import pl.touk.nussknacker.engine.api.runtimecontext.EngineRuntimeContext`
`5`	`5`
`6`	`6`	`trait FlinkEngineRuntimeContext extends EngineRuntimeContext {`
`7`		`- def runtimeContext: RuntimeContext`
	`7`	`+ def runtimeContext: FlinkEngineContext`
`8`	`8`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`package pl.touk.nussknacker.engine.process`
`2`	`2`
`3`	`3`	`import org.apache.flink.api.common.functions.{OpenContext, RichFunction}`
	`4`	`+import pl.touk.nussknacker.engine.flink.api.RuntimeCtx`
`4`	`5`	`import pl.touk.nussknacker.engine.graph.node.NodeData`
`5`	`6`	`import pl.touk.nussknacker.engine.process.compiler.FlinkProcessCompilerData`
`6`	`7`	`import pl.touk.nussknacker.engine.process.exception.FlinkExceptionHandler`
`@@ -45,7 +46,7 @@ trait ExceptionHandlerFunction extends RichFunction {`
`45`	`46`
`46`	`47`	`override def open(openContext: OpenContext): Unit = {`
`47`	`48`	`compilerData = compilerDataForClassloader(getRuntimeContext.getUserCodeClassLoader)`
`48`		`- exceptionHandler = compilerData.prepareExceptionHandler(getRuntimeContext)`
	`49`	`+ exceptionHandler = compilerData.prepareExceptionHandler(RuntimeCtx(getRuntimeContext))`
`49`	`50`	`}`
`50`	`51`
`51`	`52`	`}`