wip

TouK · Apr 9, 2024 · 22bba05 · 22bba05
1 parent a158081
commit 22bba05
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 9 deletions.
diff --git a/.../nussknacker/engine/management/sample/source/GenericSourceWithCustomVariablesSample.scala b/.../nussknacker/engine/management/sample/source/GenericSourceWithCustomVariablesSample.scala
@@ -91,10 +91,7 @@ object GenericSourceWithCustomVariablesSample
       timestampAssigner = None,
       returnType = Typed[ProcessingType],
       customContextInitializer = Some(customContextInitializer)
-    )(TypeInformation.of(classOf[ProcessingType]))
-      with TestDataGenerator
-      with FlinkSourceTestSupport[ProcessingType]
-      with CustomContextInitializerSource[ProcessingType] {
+    )(TypeInformation.of(classOf[ProcessingType])) with TestDataGenerator with FlinkSourceTestSupport[ProcessingType] {
 
       override def generateTestData(size: Int): TestData = TestData(
         elementsValue.map(el => TestRecord(Json.fromString(el)))
@@ -106,9 +103,6 @@ object GenericSourceWithCustomVariablesSample
       override def timestampAssignerForTest: Option[TimestampWatermarkHandler[String]] = timestampAssigner
 
       override def typeInformation: TypeInformation[ProcessingType] = TypeInformation.of(classOf[String])
-
-      // TODO: get rid of get
-      override def contextInitializer: ContextInitializer[ProcessingType] = customContextInitializer.get
     }
   }
 

diff --git a/...t/src/main/scala/pl/touk/nussknacker/engine/flink/util/test/FlinkTestScenarioRunner.scala b/...t/src/main/scala/pl/touk/nussknacker/engine/flink/util/test/FlinkTestScenarioRunner.scala
@@ -95,8 +95,8 @@ class FlinkTestScenarioRunner(
   def runWithData[I: ClassTag, R](
       scenario: CanonicalProcess,
       data: List[I],
-      boundedness: Boundedness = Boundedness.CONTINUOUS_UNBOUNDED,
-      flinkExecutionMode: RuntimeExecutionMode = RuntimeExecutionMode.AUTOMATIC
+      boundedness: Boundedness,
+      flinkExecutionMode: RuntimeExecutionMode
   ): RunnerListResult[R] = {
     implicit val typeInf: TypeInformation[I] =
       TypeInformation.of(implicitly[ClassTag[I]].runtimeClass.asInstanceOf[Class[I]])