correct tests

designer/server/src/main/scala/pl/touk/nussknacker/ui/definition/AlignedComponentsDefinitionProvider.scala

@@ -59,6 +59,7 @@ object AlignedComponentsDefinitionProvider {
       new BuiltInComponentsDefinitionsPreparer(designerModelData.modelData.componentsUiConfig),
       new FragmentComponentDefinitionExtractor(
         designerModelData.modelData.modelClassLoader.classLoader,
+        designerModelData.modelData.modelDefinitionWithClasses.classDefinitions.all,
         designerModelData.modelData.componentsUiConfig.groupName,
         designerModelData.modelData.determineDesignerWideId
       ),

designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/DefinitionsServiceSpec.scala

@@ -296,6 +296,7 @@ class DefinitionsServiceSpec extends AnyFunSuite with Matchers with PatientScala
       new BuiltInComponentsDefinitionsPreparer(ComponentsUiConfigParser.parse(model.modelConfig)),
       new FragmentComponentDefinitionExtractor(
         getClass.getClassLoader,
+        model.modelDefinitionWithClasses.classDefinitions.all,
         Some(_),
         DesignerWideComponentId.default(processingType.stringify, _)
       ),

designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/EdgeTypesPreparerTest.scala

@@ -22,6 +22,7 @@ class EdgeTypesPreparerTest extends AnyFunSuite with Matchers with ValidatedValu
   test("return edge types for fragment, filters, switches and components with multiple inputs") {
     val sampleFragmentDef = new FragmentComponentDefinitionExtractor(
       getClass.getClassLoader,
+      Set.empty,
       Some(_),
       DesignerWideComponentId.default(Streaming.stringify, _)
     )

designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/ComponentGroupsPreparerSpec.scala

@@ -190,6 +190,7 @@ class ComponentGroupsPreparerSpec
       new BuiltInComponentsDefinitionsPreparer(new ComponentsUiConfig(Map.empty, groupNameMapping)),
       new FragmentComponentDefinitionExtractor(
         getClass.getClassLoader,
+        Set.empty,
         Some(_),
         DesignerWideComponentId.default("Streaming", _)
       ),

designer/server/src/test/scala/pl/touk/nussknacker/ui/definition/component/ComponentsUsageHelperTest.scala

@@ -176,7 +176,7 @@ class ComponentsUsageHelperTest extends AnyFunSuite with Matchers with TableDriv
 
     val alignedComponentsDefinitionProvider = new AlignedComponentsDefinitionProvider(
       new BuiltInComponentsDefinitionsPreparer(new ComponentsUiConfig(Map.empty, Map.empty)),
-      new FragmentComponentDefinitionExtractor(getClass.getClassLoader, Some(_), determineDesignerWideId),
+      new FragmentComponentDefinitionExtractor(getClass.getClassLoader, Set.empty, Some(_), determineDesignerWideId),
       modelDefinition,
       ProcessingMode.UnboundedStream
     )

engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkProcessCompilerDataFactory.scala

@@ -9,7 +9,7 @@ import pl.touk.nussknacker.engine.api.process.{ComponentUseCase, ProcessConfigCr
 import pl.touk.nussknacker.engine.api.{JobData, MetaData, ProcessListener, ProcessVersion}
 import pl.touk.nussknacker.engine.compile._
 import pl.touk.nussknacker.engine.compile.nodecompilation.LazyParameterCreationStrategy
-import pl.touk.nussknacker.engine.definition.clazz.ClassDefinitionSet
+import pl.touk.nussknacker.engine.definition.clazz.{ClassDefinition, ClassDefinitionSet}
 import pl.touk.nussknacker.engine.definition.globalvariables.ExpressionConfigDefinition
 import pl.touk.nussknacker.engine.definition.model.{ModelDefinition, ModelDefinitionWithClasses}
 import pl.touk.nussknacker.engine.dict.DictServicesFactoryLoader
@@ -137,13 +137,18 @@ class FlinkProcessCompilerDataFactory(
       modelDefinitionWithTypes.modelDefinition.expressionConfig,
       modelDefinitionWithTypes.classDefinitions
     )
-    val adjustedDefinitions = adjustDefinitions(modelDefinitionWithTypes.modelDefinition, definitionContext)
+    val adjustedDefinitions = adjustDefinitions(
+      modelDefinitionWithTypes.modelDefinition,
+      definitionContext,
+      modelDefinitionWithTypes.classDefinitions.all
+    )
     (ModelDefinitionWithClasses(adjustedDefinitions), dictRegistry)
   }
 
   protected def adjustDefinitions(
       originalModelDefinition: ModelDefinition,
-      definitionContext: ComponentDefinitionContext
+      definitionContext: ComponentDefinitionContext,
+      classDefinitions: Set[ClassDefinition]
   ): ModelDefinition = originalModelDefinition
 
   private def loadDictRegistry(userCodeClassLoader: ClassLoader) = {

engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/StubbedFlinkProcessCompilerDataFactory.scala

@@ -10,6 +10,7 @@ import pl.touk.nussknacker.engine.api.process.{ComponentUseCase, ProcessConfigCr
 import pl.touk.nussknacker.engine.api.typed.ReturningType
 import pl.touk.nussknacker.engine.api.typed.typing.{TypingResult, Unknown}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
+import pl.touk.nussknacker.engine.definition.clazz.ClassDefinition
 import pl.touk.nussknacker.engine.definition.component.dynamic.DynamicComponentDefinitionWithImplementation
 import pl.touk.nussknacker.engine.definition.component.methodbased.MethodBasedComponentDefinitionWithImplementation
 import pl.touk.nussknacker.engine.definition.component.{
@@ -42,7 +43,8 @@ abstract class StubbedFlinkProcessCompilerDataFactory(
 
   override protected def adjustDefinitions(
       originalModelDefinition: ModelDefinition,
-      definitionContext: ComponentDefinitionContext
+      definitionContext: ComponentDefinitionContext,
+      classDefinitions: Set[ClassDefinition]
   ): ModelDefinition = {
     val usedSourceIds = process.allStartNodes
       .map(_.head.data)
@@ -61,7 +63,8 @@ abstract class StubbedFlinkProcessCompilerDataFactory(
     }
 
     val fragmentParametersDefinitionExtractor = new FragmentParametersDefinitionExtractor(
-      definitionContext.userCodeClassLoader
+      definitionContext.userCodeClassLoader,
+      classDefinitions
     )
     val fragmentSourceDefinitionPreparer = new StubbedFragmentSourceDefinitionPreparer(
       fragmentParametersDefinitionExtractor

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

@@ -53,7 +53,7 @@ object ProcessCompilerData {
     // for testing environment it's important to take classloader from user jar
     val nodeCompiler = new NodeCompiler(
       definitionWithTypes.modelDefinition,
-      new FragmentParametersDefinitionExtractor(userCodeClassLoader),
+      new FragmentParametersDefinitionExtractor(userCodeClassLoader, definitionWithTypes.classDefinitions.all),
       expressionCompiler,
       userCodeClassLoader,
       listeners,

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/definition/fragment/FragmentComponentDefinitionExtractor.scala

@@ -10,6 +10,7 @@ import pl.touk.nussknacker.engine.api.component.{
 }
 import pl.touk.nussknacker.engine.api.{FragmentSpecificData, NodeId}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
+import pl.touk.nussknacker.engine.definition.clazz.ClassDefinition
 import pl.touk.nussknacker.engine.definition.component.{
   ComponentDefinitionWithImplementation,
   ComponentImplementationInvoker
@@ -18,11 +19,12 @@ import pl.touk.nussknacker.engine.util.MetaDataExtractor
 
 class FragmentComponentDefinitionExtractor(
     classLoader: ClassLoader,
+    classDefinitions: Set[ClassDefinition],
     translateGroupName: ComponentGroupName => Option[ComponentGroupName],
     determineDesignerWideId: ComponentId => DesignerWideComponentId
 ) {
 
-  val parametersExtractor = new FragmentParametersDefinitionExtractor(classLoader)
+  val parametersExtractor = new FragmentParametersDefinitionExtractor(classLoader, classDefinitions)
 
   def extractFragmentComponentDefinition(
       fragment: CanonicalProcess,

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/definition/fragment/FragmentParametersDefinitionExtractor.scala

@@ -5,6 +5,7 @@ import cats.data.Validated.{Invalid, Valid}
 import cats.data.{Writer, WriterT}
 import cats.implicits.{catsKernelStdMonoidForList, toTraverseOps}
 import cats.instances.list._
+import org.apache.commons.lang3.ClassUtils
 import pl.touk.nussknacker.engine.api.NodeId
 import pl.touk.nussknacker.engine.api.component.ParameterConfig
 import pl.touk.nussknacker.engine.api.context.PartSubGraphCompilationError
@@ -60,7 +61,7 @@ class FragmentParametersDefinitionExtractor(
   ): WriterT[Id, List[PartSubGraphCompilationError], List[Id[Parameter]]] = {
     parameters
       .map(p =>
-        getParamTypingResultV2(p)
+        getParamTypingResult(p)
           .mapBoth { (written, typ) =>
             val param = toParameter(typ, p)
             (written ++ param.written, param.value)
@@ -138,13 +139,16 @@ class FragmentParametersDefinitionExtractor(
   }
 
   private def parseClassNameToTypingResult(className: String): Try[TypingResult] = {
+    /*
+      TODO: Write this parser in a way that handles arbitrary depth expressions
+            One should not use regexes for doing so and rather build AST
+     */
     def resolveInnerClass(simpleClassName: String): TypingResult =
       classDefinitions
         .find(classDefinition => classDefinition.clazzName.display == simpleClassName)
         .fold(
-          throw new ClassNotFoundException(
-            s"Class $simpleClassName was not found in the class definitions set: ${classDefinitions.map(_.clazzName.display)}"
-          )
+          // This is fallback - it may be removed and `ClassNotFound` exception may be thrown here after cleaning up the mess with `FragmentClazzRef` class
+          Typed(ClassUtils.getClass(classLoader, simpleClassName))
         ) { classDefinition =>
           classDefinition.clazzName
         }
@@ -168,7 +172,7 @@ class FragmentParametersDefinitionExtractor(
     })
   }
 
-  private def getParamTypingResultV2(
+  private def getParamTypingResult(
       fragmentParameter: FragmentParameter
   )(implicit nodeId: NodeId): Writer[List[PartSubGraphCompilationError], TypingResult] =
     parseClassNameToTypingResult(
@@ -183,19 +187,4 @@ class FragmentParametersDefinitionExtractor(
           )
       )
 
-  private def getParamTypingResult(
-      fragmentParameter: FragmentParameter
-  )(implicit nodeId: NodeId): Writer[List[PartSubGraphCompilationError], TypingResult] =
-    fragmentParameter.typ
-      .toRuntimeClass(classLoader)
-      .map(Typed(_))
-      .map(Writer.value[List[PartSubGraphCompilationError], TypingResult])
-      .getOrElse(
-        Writer
-          .value[List[PartSubGraphCompilationError], TypingResult](Unknown)
-          .tell(
-            List(FragmentParamClassLoadError(fragmentParameter.name, fragmentParameter.typ.refClazzName, nodeId.id))
-          )
-      )
-
 }

utils/flink-components-testkit/src/main/scala/pl/touk/nussknacker/engine/flink/util/test/FlinkProcessCompilerDataFactoryWithTestComponents.scala

@@ -8,6 +8,7 @@ import pl.touk.nussknacker.engine.api._
 import pl.touk.nussknacker.engine.api.component.{ComponentAdditionalConfig, DesignerWideComponentId}
 import pl.touk.nussknacker.engine.api.namespaces.NamingStrategy
 import pl.touk.nussknacker.engine.api.process._
+import pl.touk.nussknacker.engine.definition.clazz.ClassDefinition
 import pl.touk.nussknacker.engine.definition.component.ComponentDefinitionWithImplementation
 import pl.touk.nussknacker.engine.definition.globalvariables.GlobalVariableDefinitionWithImplementation
 import pl.touk.nussknacker.engine.definition.model.ModelDefinition
@@ -61,7 +62,8 @@ object FlinkProcessCompilerDataFactoryWithTestComponents {
 
       override protected def adjustDefinitions(
           originalModelDefinition: ModelDefinition,
-          definitionContext: ComponentDefinitionContext
+          definitionContext: ComponentDefinitionContext,
+          classDefinitions: Set[ClassDefinition]
       ): ModelDefinition = {
         val testComponents =
           ComponentDefinitionWithImplementation.forList(