Use RichFunction.open(OpenContext) instead of `RichFunction.open(Co…

…nfiguration)` (#7347)
TouK · Dec 17, 2024 · d0f0d3e · d0f0d3e
1 parent 7aa25fb
commit d0f0d3e
Show file tree

Hide file tree

Showing 22 changed files with 57 additions and 73 deletions.
diff --git a/docs/MigrationGuide.md b/docs/MigrationGuide.md
@@ -26,6 +26,8 @@ To see the biggest differences please consult the [changelog](Changelog.md).
             * `def actionTooltips(processStatus: ProcessStatus): Map[ScenarioActionName, String]` - allows to define custom tooltips for actions, if not defined the default is still used
         * modified method:
             * `def statusActions(processStatus: ProcessStatus): List[ScenarioActionName]` - changed argument, to include information about latest and deployed versions
+* [#7347](https://github.com/TouK/nussknacker/pull/7347) All calls to `org.apache.flink.api.common.functions.RichFunction.open(Configuration)`,
+  which is deprecated, were replaced with calls to `org.apache.flink.api.common.functions.RichFunction.open(OpenContext)`
 
 ## In version 1.18.0
 

diff --git a/...api/src/main/scala/pl/touk/nussknacker/engine/flink/api/datastream/StatefulFunction.scala b/...api/src/main/scala/pl/touk/nussknacker/engine/flink/api/datastream/StatefulFunction.scala
@@ -18,10 +18,9 @@
 package pl.touk.nussknacker.engine.flink.api.datastream
 
 import org.apache.flink.annotation.Public
-import org.apache.flink.api.common.functions.RichFunction
+import org.apache.flink.api.common.functions.{OpenContext, RichFunction}
 import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
 import org.apache.flink.api.common.typeutils.TypeSerializer
-import org.apache.flink.configuration.Configuration
 
 /**
  * Trait implementing the functionality necessary to apply stateful functions in RichFunctions
@@ -44,7 +43,7 @@ trait StatefulFunction[I, O, S] extends RichFunction {
     o
   }
 
-  override def open(c: Configuration) = {
+  override def open(openContext: OpenContext): Unit = {
     val info = new ValueStateDescriptor[S]("state", stateSerializer)
     state = getRuntimeContext().getState(info)
   }

diff --git a/.../src/main/scala/pl/touk/nussknacker/engine/flink/api/exception/WithExceptionHandler.scala b/.../src/main/scala/pl/touk/nussknacker/engine/flink/api/exception/WithExceptionHandler.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.api.exception
 
-import org.apache.flink.api.common.functions.{RichFunction, RuntimeContext}
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RichFunction, RuntimeContext}
 
 /**
   * Helper for using exception handler.
@@ -15,7 +14,7 @@ trait WithExceptionHandler {
 
   protected var exceptionHandler: ExceptionHandler = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     exceptionHandler = exceptionHandlerPreparer(getRuntimeContext)
   }
 

diff --git a/...scala/pl/touk/nussknacker/engine/flink/api/process/FlinkContextInitializingFunction.scala b/...scala/pl/touk/nussknacker/engine/flink/api/process/FlinkContextInitializingFunction.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.api.process
 
-import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RichMapFunction, RuntimeContext}
 import pl.touk.nussknacker.engine.api.Context
 import pl.touk.nussknacker.engine.api.process.{ContextInitializer, ContextInitializingFunction}
 import pl.touk.nussknacker.engine.api.runtimecontext.EngineRuntimeContext
@@ -14,7 +13,7 @@ class FlinkContextInitializingFunction[Raw](
 
   private var initializingStrategy: ContextInitializingFunction[Raw] = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     val contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId)
     initializingStrategy = contextInitializer.initContext(contextIdGenerator)
   }

diff --git a/...scala/pl/touk/nussknacker/engine/flink/api/process/FlinkLazyParameterFunctionHelper.scala b/...scala/pl/touk/nussknacker/engine/flink/api/process/FlinkLazyParameterFunctionHelper.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.api.process
 
 import org.apache.flink.api.common.functions._
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.util.Collector
 import pl.touk.nussknacker.engine.api.LazyParameter.Evaluate
 import pl.touk.nussknacker.engine.api._
@@ -89,8 +88,8 @@ trait OneParamLazyParameterFunction[T <: AnyRef] extends LazyParameterInterprete
   protected def evaluateParameter(ctx: Context): T =
     _evaluateParameter(ctx)
 
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    super.open(openContext)
     _evaluateParameter = toEvaluateFunctionConverter.toEvaluateFunction(parameter)
   }
 
@@ -117,7 +116,7 @@ trait LazyParameterInterpreterFunction { self: RichFunction =>
   }
 
   // TODO: how can we make sure this will invoke super.open(...) (can't do it directly...)
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     toEvaluateFunctionConverter = lazyParameterHelper.createInterpreter(getRuntimeContext)
     exceptionHandler = lazyParameterHelper.exceptionHandler(getRuntimeContext)
   }

diff --git a/...i/src/main/scala/pl/touk/nussknacker/engine/flink/api/process/RichLifecycleFunction.scala b/...i/src/main/scala/pl/touk/nussknacker/engine/flink/api/process/RichLifecycleFunction.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.api.process
 
-import org.apache.flink.api.common.functions.{AbstractRichFunction, MapFunction, RuntimeContext}
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{AbstractRichFunction, MapFunction, OpenContext, RuntimeContext}
 import pl.touk.nussknacker.engine.api.Lifecycle
 import pl.touk.nussknacker.engine.api.runtimecontext.EngineRuntimeContext
 
@@ -11,7 +10,7 @@ abstract class RichLifecycleFunction extends AbstractRichFunction {
 
   protected val convertToEngineRuntimeContext: RuntimeContext => EngineRuntimeContext
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     lifecycle.open(convertToEngineRuntimeContext(getRuntimeContext))
   }
 

diff --git a/...ents-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala b/...ents-utils/src/main/scala/pl/touk/nussknacker/engine/flink/api/state/EvictableState.scala
@@ -1,8 +1,7 @@
 package pl.touk.nussknacker.engine.flink.api.state
 
-import org.apache.flink.api.common.functions.RichFunction
+import org.apache.flink.api.common.functions.{OpenContext, RichFunction}
 import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.TimerService
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction
 import org.apache.flink.streaming.api.functions.co.CoProcessFunction
@@ -18,8 +17,8 @@ abstract class EvictableStateFunction[In, Out, StateType] extends KeyedProcessFu
 
   protected def stateDescriptor: ValueStateDescriptor[StateType]
 
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    super.open(openContext)
     lastEventTimeForKey = getRuntimeContext.getState[java.lang.Long](
       new ValueStateDescriptor[java.lang.Long]("timers", classOf[java.lang.Long])
     )
@@ -91,7 +90,7 @@ trait LatelyEvictableStateFunctionMixin[StateType] extends RichFunction with Sta
   @transient
   protected var state: ValueState[StateType] = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     latestEvictionTimeForKey = getRuntimeContext.getState[java.lang.Long](
       new ValueStateDescriptor[java.lang.Long]("timers", classOf[java.lang.Long])
     )

diff --git a/...n/scala/pl/touk/nussknacker/engine/flink/util/function/CoProcessFunctionInterceptor.scala b/...n/scala/pl/touk/nussknacker/engine/flink/util/function/CoProcessFunctionInterceptor.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.util.function
 
-import org.apache.flink.api.common.functions.RuntimeContext
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RuntimeContext}
 import org.apache.flink.streaming.api.functions.co.CoProcessFunction
 import org.apache.flink.util.Collector
 
@@ -12,8 +11,8 @@ import org.apache.flink.util.Collector
 abstract class CoProcessFunctionInterceptor[IN1, IN2, OUT](underlying: CoProcessFunction[IN1, IN2, OUT])
     extends CoProcessFunction[IN1, IN2, OUT] {
 
-  override def open(parameters: Configuration): Unit = {
-    underlying.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    underlying.open(openContext)
   }
 
   override def setRuntimeContext(ctx: RuntimeContext): Unit = {

diff --git a/...ain/scala/pl/touk/nussknacker/engine/flink/util/function/ProcessFunctionInterceptor.scala b/...ain/scala/pl/touk/nussknacker/engine/flink/util/function/ProcessFunctionInterceptor.scala
@@ -1,16 +1,14 @@
 package pl.touk.nussknacker.engine.flink.util.function
 
-import org.apache.flink.api.common.functions.RuntimeContext
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RuntimeContext}
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction
-import org.apache.flink.streaming.api.functions.co.CoProcessFunction
 import org.apache.flink.util.Collector
 
 abstract class ProcessFunctionInterceptor[IN, OUT](underlying: KeyedProcessFunction[String, IN, OUT])
     extends KeyedProcessFunction[String, IN, OUT] {
 
-  override def open(parameters: Configuration): Unit = {
-    underlying.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    underlying.open(openContext)
   }
 
   override def setRuntimeContext(ctx: RuntimeContext): Unit = {

diff --git a/...link/util/transformer/aggregate/EmitExtraWindowWhenNoDataTumblingAggregatorFunction.scala b/...link/util/transformer/aggregate/EmitExtraWindowWhenNoDataTumblingAggregatorFunction.scala
@@ -1,9 +1,8 @@
 package pl.touk.nussknacker.engine.flink.util.transformer.aggregate
 
-import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.api.common.functions.{OpenContext, RuntimeContext}
 import org.apache.flink.api.common.state.ValueState
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.TimerService
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction
 import org.apache.flink.util.Collector
@@ -47,7 +46,7 @@ class EmitExtraWindowWhenNoDataTumblingAggregatorFunction[MapT[K, V]](
   @transient
   private var contextIdGenerator: ContextIdGenerator = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     state = getRuntimeContext.getState(stateDescriptor)
     contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId.id)
   }

diff --git a/...knacker/engine/flink/util/transformer/aggregate/EmitWhenEventLeftAggregatorFunction.scala b/...knacker/engine/flink/util/transformer/aggregate/EmitWhenEventLeftAggregatorFunction.scala
@@ -1,8 +1,7 @@
 package pl.touk.nussknacker.engine.flink.util.transformer.aggregate
 
-import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.api.common.functions.{OpenContext, RuntimeContext}
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.TimerService
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction
 import org.apache.flink.util.Collector
@@ -42,8 +41,8 @@ class EmitWhenEventLeftAggregatorFunction[MapT[K, V]](
   type FlinkOnTimerCtx =
     KeyedProcessFunction[String, ValueWithContext[StringKeyedValue[AnyRef]], ValueWithContext[AnyRef]]#OnTimerContext
 
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    super.open(openContext)
     contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId.id)
   }
 

diff --git a/...l/touk/nussknacker/engine/flink/util/transformer/aggregate/EnrichingWithKeyFunction.scala b/...l/touk/nussknacker/engine/flink/util/transformer/aggregate/EnrichingWithKeyFunction.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.flink.util.transformer.aggregate
 
-import org.apache.flink.api.common.functions.RuntimeContext
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RuntimeContext}
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow
 import org.apache.flink.util.Collector
@@ -18,7 +17,7 @@ class EnrichingWithKeyFunction(convertToEngineRuntimeContext: RuntimeContext =>
   @transient
   private var contextIdGenerator: ContextIdGenerator = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId)
   }
 

diff --git a/...uk/nussknacker/engine/flink/util/transformer/aggregate/OnEventTriggerWindowOperator.scala b/...uk/nussknacker/engine/flink/util/transformer/aggregate/OnEventTriggerWindowOperator.scala
@@ -1,9 +1,8 @@
 package pl.touk.nussknacker.engine.flink.util.transformer.aggregate
 
 import com.github.ghik.silencer.silent
-import org.apache.flink.api.common.functions.{AggregateFunction, RuntimeContext}
+import org.apache.flink.api.common.functions.{AggregateFunction, OpenContext, RuntimeContext}
 import org.apache.flink.api.common.state.AggregatingStateDescriptor
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.datastream.{KeyedStream, SingleOutputStreamOperator}
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
 import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner
@@ -102,7 +101,7 @@ private class ValueEmittingWindowFunction(
   @transient
   private var contextIdGenerator: ContextIdGenerator = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId)
   }
 

diff --git a/...e/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/DelayTransformer.scala b/...e/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/DelayTransformer.scala
@@ -1,9 +1,9 @@
 package pl.touk.nussknacker.engine.flink.util.transformer
 
+import org.apache.flink.api.common.functions.OpenContext
 import org.apache.flink.api.common.state.{MapState, MapStateDescriptor}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.typeutils.ListTypeInfo
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.datastream.DataStream
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction
 import org.apache.flink.util.Collector
@@ -75,7 +75,7 @@ class DelayFunction(nodeCtx: FlinkCustomNodeContext, delay: Duration)
 
   @transient private var state: MapState[Long, java.util.List[api.Context]] = _
 
-  override def open(config: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     state = getRuntimeContext.getMapState(descriptor)
   }
 

diff --git a/...in/scala/pl/touk/nussknacker/engine/flink/util/transformer/PreviousValueTransformer.scala b/...in/scala/pl/touk/nussknacker/engine/flink/util/transformer/PreviousValueTransformer.scala
@@ -1,9 +1,8 @@
 package pl.touk.nussknacker.engine.flink.util.transformer
 
-import org.apache.flink.api.common.functions.RichFlatMapFunction
+import org.apache.flink.api.common.functions.{OpenContext, RichFlatMapFunction}
 import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.datastream.DataStream
 import org.apache.flink.util.Collector
 import pl.touk.nussknacker.engine.api._
@@ -46,8 +45,8 @@ case object PreviousValueTransformer extends CustomStreamTransformer with Explic
 
     private[this] var state: ValueState[Value] = _
 
-    override def open(c: Configuration): Unit = {
-      super.open(c)
+    override def open(openContext: OpenContext): Unit = {
+      super.open(openContext)
       val info = new ValueStateDescriptor[Value]("state", typeInformation)
       state = getRuntimeContext.getState(info)
     }

diff --git a/...le/src/main/scala/pl/touk/nussknacker/engine/flink/table/aggregate/TableAggregation.scala b/...le/src/main/scala/pl/touk/nussknacker/engine/flink/table/aggregate/TableAggregation.scala
@@ -1,8 +1,7 @@
 package pl.touk.nussknacker.engine.flink.table.aggregate
 
-import org.apache.flink.api.common.functions.{FlatMapFunction, RuntimeContext}
+import org.apache.flink.api.common.functions.{FlatMapFunction, OpenContext, RuntimeContext}
 import org.apache.flink.api.common.typeinfo.{TypeInformation, Types}
-import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.datastream.DataStream
 import org.apache.flink.streaming.api.functions.ProcessFunction
 import org.apache.flink.table.api.Expressions.{$, call}
@@ -116,7 +115,7 @@ class TableAggregation(
     @transient
     private var contextIdGenerator: ContextIdGenerator = _
 
-    override def open(configuration: Configuration): Unit = {
+    override def open(openContext: OpenContext): Unit = {
       contextIdGenerator = convertToEngineRuntimeContext(getRuntimeContext).contextIdGenerator(nodeId.toString)
     }
 

diff --git a/...link/executor/src/main/scala/pl/touk/nussknacker/engine/process/ProcessPartFunction.scala b/...link/executor/src/main/scala/pl/touk/nussknacker/engine/process/ProcessPartFunction.scala
@@ -1,7 +1,6 @@
 package pl.touk.nussknacker.engine.process
 
-import org.apache.flink.api.common.functions.RichFunction
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.{OpenContext, RichFunction}
 import pl.touk.nussknacker.engine.graph.node.NodeData
 import pl.touk.nussknacker.engine.process.compiler.FlinkProcessCompilerData
 import pl.touk.nussknacker.engine.process.exception.FlinkExceptionHandler
@@ -22,8 +21,8 @@ trait ProcessPartFunction extends ExceptionHandlerFunction {
     }
   }
 
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    super.open(openContext)
     compilerData.open(getRuntimeContext, nodesUsed)
   }
 
@@ -46,7 +45,7 @@ trait ExceptionHandlerFunction extends RichFunction {
     }
   }
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     exceptionHandler = compilerData.prepareExceptionHandler(getRuntimeContext)
   }
 

diff --git a/...main/scala/pl/touk/nussknacker/engine/process/registrar/AsyncInterpretationFunction.scala b/...main/scala/pl/touk/nussknacker/engine/process/registrar/AsyncInterpretationFunction.scala
@@ -3,7 +3,7 @@ package pl.touk.nussknacker.engine.process.registrar
 import cats.effect.IO
 import cats.effect.unsafe.IORuntime
 import com.typesafe.scalalogging.LazyLogging
-import org.apache.flink.configuration.Configuration
+import org.apache.flink.api.common.functions.OpenContext
 import org.apache.flink.streaming.api.functions.async.{ResultFuture, RichAsyncFunction}
 import pl.touk.nussknacker.engine.InterpretationResult
 import pl.touk.nussknacker.engine.Interpreter.FutureShape
@@ -35,8 +35,8 @@ private[registrar] class AsyncInterpretationFunction(
 
   private var serviceExecutionContext: ServiceExecutionContext = _
 
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
+  override def open(openContext: OpenContext): Unit = {
+    super.open(openContext)
 
     getRuntimeContext.registerUserCodeClassLoaderReleaseHookIfAbsent(
       "closeAsyncExecutionContext",

diff --git a/...r/src/main/scala/pl/touk/nussknacker/engine/process/registrar/SourceMetricsFunction.scala b/...r/src/main/scala/pl/touk/nussknacker/engine/process/registrar/SourceMetricsFunction.scala
@@ -1,6 +1,6 @@
 package pl.touk.nussknacker.engine.process.registrar
 
-import org.apache.flink.configuration.Configuration
+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.api.process.ComponentUseCase
@@ -12,7 +12,7 @@ private[registrar] class SourceMetricsFunction[T](sourceId: String, componentUse
 
   @transient private var metrics: OneSourceMetrics = _
 
-  override def open(parameters: Configuration): Unit = {
+  override def open(openContext: OpenContext): Unit = {
     metrics = new OneSourceMetrics(sourceId)
     val metricsProvider = createMetricsProvider(componentUseCase, getRuntimeContext)
     metrics.registerOwnMetrics(metricsProvider)