tests fix

Author: arkadius

engine/flink/components/base-tests/src/test/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/TransformersTest.scala

@@ -494,7 +494,7 @@ class TransformersTest extends AnyFunSuite with FlinkSpec with Matchers with Ins
       val ex = the[IllegalArgumentException] thrownBy {
         runScenario(model, resolvedScenario)
       }
-      ex should have message "Compilation errors: ExpressionParserCompilationError(Unresolved reference 'input',inputVarAccessTest,Some(ParameterName($expression)),#input,None)"
+      ex should have message "Compilation errors: ExpressionParserCompilationError(Unresolved reference 'input',inputVarAccessTest,Some($expression),#input,None)"
     }
 
   }

engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/functional/CustomNodeProcessSpec.scala

@@ -261,7 +261,7 @@ class CustomNodeProcessSpec extends AnyFunSuite with Matchers with ProcessTestHe
     val thrown = the[IllegalArgumentException] thrownBy processInvoker.invokeWithSampleData(process, data)
 
     thrown.getMessage should startWith(
-      "Compilation errors: ExpressionParserCompilationError(Unresolved reference 'input',proc2,Some(ParameterName(all)),#input.id,None)"
+      "Compilation errors: ExpressionParserCompilationError(Unresolved reference 'input',proc2,Some(all),#input.id,None)"
     )
   }
 
@@ -276,7 +276,7 @@ class CustomNodeProcessSpec extends AnyFunSuite with Matchers with ProcessTestHe
 
     val thrown = the[IllegalArgumentException] thrownBy processInvoker.invokeWithSampleData(process, List.empty)
 
-    thrown.getMessage shouldBe s"Compilation errors: ExpressionParserCompilationError(There is no property 'value999' in type: SimpleRecord,delta,Some(ParameterName($$expression)),#outRec.record.value999 > #outRec.previous + 5,None)"
+    thrown.getMessage shouldBe s"Compilation errors: ExpressionParserCompilationError(There is no property 'value999' in type: SimpleRecord,delta,Some($$expression),#outRec.record.value999 > #outRec.previous + 5,None)"
   }
 
   test("should evaluate blank expression used in lazy parameter as a null") {

engine/flink/schemed-kafka-components-utils/src/test/scala/pl/touk/nussknacker/engine/schemedkafka/KafkaAvroPayloadIntegrationSpec.scala

@@ -96,7 +96,7 @@ class KafkaAvroPayloadIntegrationSpec extends KafkaAvroSpecMixin with BeforeAndA
       runAndVerifyResultSingleEvent(process, topicConfig, "fooBar", "fooBar")
     }.getMessage
 
-    message should include("InvalidPropertyFixedValue(ParameterName(Topic),None,'invalid-topic',")
+    message should include("InvalidPropertyFixedValue(Topic,None,'invalid-topic',")
   }
 
   test("should handle null value for mandatory parameter") {
@@ -110,7 +110,7 @@ class KafkaAvroPayloadIntegrationSpec extends KafkaAvroSpecMixin with BeforeAndA
     }.getMessage
 
     message should include(
-      "EmptyMandatoryParameter(This field is mandatory and can not be empty,Please fill field for this parameter,ParameterName(Topic),start"
+      "EmptyMandatoryParameter(This field is mandatory and can not be empty,Please fill field for this parameter,Topic,start"
     )
   }
 

engine/flink/schemed-kafka-components-utils/src/test/scala/pl/touk/nussknacker/engine/schemedkafka/source/flink/DelayedUniversalKafkaSourceAvroPayloadIntegrationSpec.scala

@@ -146,7 +146,7 @@ class DelayedUniversalKafkaSourceAvroPayloadIntegrationSpec
     intercept[IllegalArgumentException] {
       runAndVerify(topicConfig, process, LongFieldV1.record)
     }.getMessage should include(
-      "LowerThanRequiredParameter(This field value has to be a number greater than or equal to 0,Please fill field with proper number,ParameterName(delayInMillis),start)"
+      "LowerThanRequiredParameter(This field value has to be a number greater than or equal to 0,Please fill field with proper number,delayInMillis,start)"
     )
   }
 

engine/flink/tests/src/test/scala/pl/touk/nussknacker/engine/flink/minicluster/scenariotesting/FlinkMiniClusterScenarioTestRunnerSpec.scala

@@ -833,7 +833,7 @@ class FlinkMiniClusterScenarioTestRunnerSpec
       }.getCause
       inside(dictEditorException) { case _: IllegalArgumentException =>
         dictEditorException.getMessage.startsWith(
-          "Compilation errors: IncompatibleParameterDefinitionModification(ParameterName(static),dictKeyWithLabel,List(SpelTemplateParameterEditor, SpelParameterEditor)"
+          "Compilation errors: IncompatibleParameterDefinitionModification(static,dictKeyWithLabel,List(SpelTemplateParameterEditor, SpelParameterEditor)"
         ) shouldBe true
       }
     }

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/NodeParametersAdjuster.scala

@@ -1,37 +1,37 @@
 package pl.touk.nussknacker.engine.compile
 
+import cats.Id
+import cats.data.{NonEmptySet, WriterT}
 import cats.data.Validated.valid
-import cats.implicits.catsSyntaxValidatedId
+import cats.implicits.toTraverseOps
+import cats.instances.set
 import com.typesafe.scalalogging.LazyLogging
 import pl.touk.nussknacker.engine.api.{JobData, NodeId}
 import pl.touk.nussknacker.engine.api.context.ProcessCompilationError.MissingParameters
-import pl.touk.nussknacker.engine.api.definition.Parameter
+import pl.touk.nussknacker.engine.api.definition.{Parameter => ParameterDefinition}
 import pl.touk.nussknacker.engine.api.parameter.ParameterName
 import pl.touk.nussknacker.engine.graph.evaluatedparam.{Parameter => NodeParameter}
+import pl.touk.nussknacker.engine.util.Implicits.RichTupleList
 
 object NodeParametersAdjuster extends LazyLogging {
 
-  def adjustNonBranchParameters(parameterDefinitions: List[Parameter], parameters: List[NodeParameter])(
+  def adjustNonBranchParameters(parameterDefinitions: List[ParameterDefinition], parameters: List[NodeParameter])(
       implicit nodeId: NodeId,
       jobData: JobData
   ): List[NodeParameter] = {
-    val definedParamNamesSet = parameterDefinitions.filter(!_.branchParam).map(_.name).toSet
-    val usedParamNamesSet    = parameters.map(_.name).toSet
-
-    checkRedundancyAndWarnIfNeeded(definedParamNamesSet, usedParamNamesSet)
-    validateMissingness(definedParamNamesSet, usedParamNamesSet)
-    // FIXME abr
-    parameters
+    val nonBranchParametersDefinition = parameterDefinitions.filter(!_.branchParam)
+    checkRedundancyAndWarnIfNeeded(nonBranchParametersDefinition, parameters)
+    addMissingNodeParameters(nonBranchParametersDefinition, parameters)
   }
 
   private def checkRedundancyAndWarnIfNeeded(
-      definedParamNamesSet: Set[ParameterName],
-      usedParamNamesSet: Set[ParameterName]
+      parameterDefinitions: List[ParameterDefinition],
+      parameters: List[NodeParameter]
   )(
       implicit nodeId: NodeId,
       jobData: JobData
   ): Unit = {
-    val redundantParams = usedParamNamesSet.diff(definedParamNamesSet)
+    val redundantParams = parameters.map(_.name).toSet.diff(parameterDefinitions.map(_.name).toSet)
     if (redundantParams.nonEmpty) {
       logger.warn(
         s"Scenario [${jobData.metaData.name}] node [$nodeId] compilation warning. " +
@@ -40,11 +40,37 @@ object NodeParametersAdjuster extends LazyLogging {
     }
   }
 
-  private def validateMissingness(definedParamNamesSet: Set[ParameterName], usedParamNamesSet: Set[ParameterName])(
-      implicit nodeId: NodeId
+  private def addMissingNodeParameters(
+      parameterDefinitions: List[ParameterDefinition],
+      parameters: List[NodeParameter]
+  )(
+      implicit nodeId: NodeId,
+      jobData: JobData
   ) = {
-    val notUsedParams = definedParamNamesSet.diff(usedParamNamesSet)
-    if (notUsedParams.nonEmpty) MissingParameters(notUsedParams).invalidNel[Unit] else valid(())
+    val parameterByName = parameters.map(p => p.name -> p).toMapCheckingDuplicates
+    val (missingParameterNames, adjustedParameters) = parameterDefinitions
+      .map { parameterDefinition =>
+        parameterByName
+          .get(parameterDefinition.name)
+          .map(parameter => WriterT.value[Id, Set[ParameterName], NodeParameter](parameter))
+          .getOrElse {
+            WriterT
+              .value[Id, Set[ParameterName], NodeParameter](
+                NodeParameter(parameterDefinition.name, parameterDefinition.finalDefaultValue)
+              )
+              .tell(Set(parameterDefinition.name))
+          }
+      }
+      .sequence
+      .run
+    if (missingParameterNames.nonEmpty) {
+      logger.warn(
+        s"Scenario [${jobData.metaData.name}] node [$nodeId] compilation warning. " +
+          s"Found missing parameters: ${missingParameterNames.toList.map(_.value).sorted.mkString(", ")}. " +
+          s"They will be recovered based on the default value from parameters definitions."
+      )
+    }
+    adjustedParameters
   }
 
 }

scenario-compiler/src/test/scala/pl/touk/nussknacker/engine/InterpreterSpec.scala

@@ -965,7 +965,7 @@ class InterpreterSpec extends AnyFunSuite with Matchers {
 
     intercept[IllegalArgumentException] {
       interpretProcess(process, Transaction())
-    }.getMessage shouldBe "Compilation errors: EmptyMandatoryParameter(This field is mandatory and can not be empty,Please fill field for this parameter,ParameterName(expression),customNode)"
+    }.getMessage shouldBe "Compilation errors: EmptyMandatoryParameter(This field is mandatory and can not be empty,Please fill field for this parameter,expression,customNode)"
   }
 
   test("not accept blank expression for not blank parameter") {
@@ -978,7 +978,7 @@ class InterpreterSpec extends AnyFunSuite with Matchers {
 
     intercept[IllegalArgumentException] {
       interpretProcess(process, Transaction())
-    }.getMessage shouldBe "Compilation errors: BlankParameter(This field value is required and can not be blank,Please fill field value for this parameter,ParameterName(expression),customNode)"
+    }.getMessage shouldBe "Compilation errors: BlankParameter(This field value is required and can not be blank,Please fill field value for this parameter,expression,customNode)"
   }
 
   test("use eager service") {

scenario-compiler/src/test/scala/pl/touk/nussknacker/engine/compile/ProcessValidatorSpec.scala

@@ -956,19 +956,6 @@ class ProcessValidatorSpec extends AnyFunSuite with Matchers with Inside with Op
     }
   }
 
-  test("validate service params") {
-    val process = ScenarioBuilder
-      .streaming("process1")
-      .source("id1", "typed-source")
-      .enricher("enricher1", "out", "withParamsService")
-      .emptySink("id2", "sink")
-
-    inside(validate(process, definitionWithTypedSource).result) {
-      case Invalid(NonEmptyList(MissingParameters(missingParam, "enricher1"), _)) =>
-        missingParam shouldBe Set(ParameterName("par1"))
-    }
-  }
-
   test("find usage of fields that does not exist in option object") {
     val process = ScenarioBuilder
       .streaming("process1")