[NU-1962] Flink test mechanism: caching flink mini cluster

build.sbt

@@ -631,7 +631,9 @@ lazy val flinkDeploymentManager = (project in flink("management"))
     componentsApi        % Provided,
     httpUtils            % Provided,
     flinkScalaUtils      % Provided,
-    flinkTestUtils       % IntegrationTest,
+    // test->test dependency is needed to load SimpleProcessConfigCreator
+    flinkExecutor        % "test,test->test",
+    flinkTestUtils       % "it,test",
     kafkaTestUtils       % "it,test"
   )
 
@@ -710,18 +712,22 @@ lazy val flinkTests = (project in flink("tests"))
     }
   )
   .dependsOn(
-    defaultModel                 % Test,
-    flinkExecutor                % Test,
-    flinkKafkaComponents         % Test,
-    flinkBaseComponents          % Test,
-    flinkBaseUnboundedComponents % Test,
-    flinkTableApiComponents      % Test,
-    flinkTestUtils               % Test,
-    kafkaTestUtils               % Test,
-    flinkComponentsTestkit       % Test,
+    defaultModel                     % Test,
+    flinkExecutor                    % Test,
+    flinkKafkaComponents             % Test,
+    flinkBaseComponents              % Test,
+    flinkBaseUnboundedComponents     % Test,
+    flinkTableApiComponents          % Test,
+    flinkTestUtils                   % Test,
+    kafkaTestUtils                   % Test,
+    flinkComponentsTestkit           % Test,
+    flinkDeploymentManager           % Test,
+    // test->test dependencies are needed to load components from these modules
+    flinkKafkaComponentsUtils        % "test->test",
+    flinkSchemedKafkaComponentsUtils % "test->test",
     // for local development
-    designer                     % Test,
-    deploymentManagerApi         % Test
+    designer                         % Test,
+    deploymentManagerApi             % Test
   )
 
 lazy val defaultModel = (project in (file("defaultModel")))
@@ -965,7 +971,7 @@ lazy val flinkSchemedKafkaComponentsUtils = (project in flink("schemed-kafka-com
     componentsUtils             % Provided,
     kafkaTestUtils              % Test,
     flinkTestUtils              % Test,
-    flinkExecutor               % Test
+    flinkExecutor               % Test,
   )
 
 lazy val flinkKafkaComponentsUtils = (project in flink("kafka-components-utils"))

designer/server/src/main/scala/pl/touk/nussknacker/ui/api/description/TestingApiEndpoints.scala

@@ -1,34 +1,33 @@
 package pl.touk.nussknacker.ui.api.description
 
+import io.circe.Encoder
 import pl.touk.nussknacker.engine.api.StreamMetaData
 import pl.touk.nussknacker.engine.api.definition.Parameter
 import pl.touk.nussknacker.engine.api.graph.{ProcessProperties, ScenarioGraph}
 import pl.touk.nussknacker.engine.api.parameter.ParameterName
 import pl.touk.nussknacker.engine.api.process.ProcessName
-import pl.touk.nussknacker.engine.api.typed.typing.Typed
+import pl.touk.nussknacker.engine.api.typed.typing._
 import pl.touk.nussknacker.engine.definition.test.TestingCapabilities
 import pl.touk.nussknacker.engine.graph.expression.Expression
 import pl.touk.nussknacker.restmodel.BaseEndpointDefinitions
 import pl.touk.nussknacker.restmodel.BaseEndpointDefinitions.SecuredEndpoint
-import pl.touk.nussknacker.restmodel.definition.{UIParameter, UISourceParameters}
+import pl.touk.nussknacker.restmodel.definition.UISourceParameters
 import pl.touk.nussknacker.restmodel.validation.ValidationResults.{NodeValidationError, NodeValidationErrorType}
-import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioNameCodec._
 import pl.touk.nussknacker.security.AuthCredentials
 import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioGraphCodec._
+import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioNameCodec._
 import pl.touk.nussknacker.ui.api.TapirCodecs.ScenarioTestingCodecs._
-import pl.touk.nussknacker.ui.definition.DefinitionsService
-import sttp.model.StatusCode.Ok
-import sttp.tapir.EndpointIO.Example
-import sttp.tapir._
-import sttp.tapir.json.circe.jsonBody
-import io.circe.Encoder
-import pl.touk.nussknacker.engine.api.typed.typing._
 import pl.touk.nussknacker.ui.api.TestingApiHttpService.Examples.{
   malformedTypingResultExample,
   noScenarioExample,
   testDataGenerationErrorExample
 }
 import pl.touk.nussknacker.ui.api.TestingApiHttpService.TestingError
+import pl.touk.nussknacker.ui.definition.DefinitionsService
+import sttp.model.StatusCode.Ok
+import sttp.tapir.EndpointIO.Example
+import sttp.tapir._
+import sttp.tapir.json.circe.jsonBody
 
 class TestingApiEndpoints(auth: EndpointInput[AuthCredentials]) extends BaseEndpointDefinitions {
   import NodesApiEndpoints.Dtos._

designer/server/src/main/scala/pl/touk/nussknacker/ui/process/periodic/legacy/db/LegacyPeriodicProcessDeploymentsTableFactory.scala

@@ -1,7 +1,11 @@
 package pl.touk.nussknacker.ui.process.periodic.legacy.db
 
 import pl.touk.nussknacker.ui.process.periodic.model.PeriodicProcessDeploymentStatus.PeriodicProcessDeploymentStatus
-import pl.touk.nussknacker.ui.process.periodic.model.{PeriodicProcessDeploymentId, PeriodicProcessDeploymentStatus, PeriodicProcessId}
+import pl.touk.nussknacker.ui.process.periodic.model.{
+  PeriodicProcessDeploymentId,
+  PeriodicProcessDeploymentStatus,
+  PeriodicProcessId
+}
 import slick.jdbc.{JdbcProfile, JdbcType}
 import slick.lifted.ProvenShape
 import slick.sql.SqlProfile.ColumnOption.NotNull

e2e-tests/src/test/scala/pl/touk/nussknacker/BatchDataGenerationSpec.scala

@@ -27,8 +27,9 @@ class BatchDataGenerationSpec
 
   private val designerServiceUrl = "http://localhost:8080"
 
-  private val liveDataGenScenarioName   = "SumTransactions-LiveData"
-  private val randomDataGenScenarioName = "SumTransactions-RandomData"
+  private val timeBasedRandomSuffix     = System.currentTimeMillis()
+  private val liveDataGenScenarioName   = s"SumTransactions-LiveData-$timeBasedRandomSuffix"
+  private val randomDataGenScenarioName = s"SumTransactions-RandomData-$timeBasedRandomSuffix"
 
   override def beforeAll(): Unit = {
     createEmptyBatchScenario(liveDataGenScenarioName, "Default")
@@ -106,7 +107,7 @@ class BatchDataGenerationSpec
            |    "nodeResults": {
            |      "sourceId": [
            |        {
-           |          "id": "SumTransactions-LiveData-sourceId-0-0",
+           |          "id": "$liveDataGenScenarioName-sourceId-0-0",
            |          "variables": {
            |            "input": {
            |              "pretty": {
@@ -122,7 +123,7 @@ class BatchDataGenerationSpec
            |      ],
            |      "end": [
            |        {
-           |          "id": "SumTransactions-LiveData-sourceId-0-0",
+           |          "id": "$liveDataGenScenarioName-sourceId-0-0",
            |          "variables": {
            |            "input": {
            |              "pretty": {
@@ -183,7 +184,7 @@ class BatchDataGenerationSpec
            |    "nodeResults": {
            |      "sourceId": [
            |        {
-           |          "id": "SumTransactions-LiveData-sourceId-0-0",
+           |          "id": "$liveDataGenScenarioName-sourceId-0-0",
            |          "variables": {
            |            "input": {
            |              "pretty": {
@@ -199,7 +200,7 @@ class BatchDataGenerationSpec
            |      ],
            |      "end": [
            |        {
-           |          "id": "SumTransactions-LiveData-sourceId-0-0",
+           |          "id": "$liveDataGenScenarioName-sourceId-0-0",
            |          "variables": {
            |            "input": {
            |              "pretty": {

engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/DevelopmentDeploymentManagerProvider.scala

@@ -4,6 +4,7 @@ import akka.actor.ActorSystem
 import cats.data.{Validated, ValidatedNel}
 import com.typesafe.config.Config
 import com.typesafe.scalalogging.LazyLogging
+import org.apache.flink.configuration.Configuration
 import pl.touk.nussknacker.development.manager.DevelopmentStateStatus._
 import pl.touk.nussknacker.engine._
 import pl.touk.nussknacker.engine.api.ProcessVersion
@@ -19,7 +20,8 @@ import pl.touk.nussknacker.engine.api.deployment.simple.{SimpleProcessStateDefin
 import pl.touk.nussknacker.engine.api.process.ProcessName
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.deployment._
-import pl.touk.nussknacker.engine.management.{FlinkProcessTestRunner, FlinkStreamingPropertiesConfig}
+import pl.touk.nussknacker.engine.management.FlinkStreamingPropertiesConfig
+import pl.touk.nussknacker.engine.management.testsmechanism.FlinkProcessTestRunner
 
 import java.net.URI
 import java.util.UUID
@@ -48,7 +50,8 @@ class DevelopmentDeploymentManager(actorSystem: ActorSystem, modelData: BaseMode
   private val memory: TrieMap[ProcessName, StatusDetails] = TrieMap[ProcessName, StatusDetails]()
   private val random                                      = new scala.util.Random()
 
-  private lazy val flinkTestRunner = new FlinkProcessTestRunner(modelData.asInvokableModelData)
+  private lazy val flinkTestRunner =
+    new FlinkProcessTestRunner(modelData.asInvokableModelData, parallelism = 1, new Configuration())
 
   implicit private class ProcessStateExpandable(processState: StatusDetails) {
 
@@ -83,7 +86,10 @@ class DevelopmentDeploymentManager(actorSystem: ActorSystem, modelData: BaseMode
     case command: DMRunOffScheduleCommand  => runOffSchedule(command)
     case _: DMMakeScenarioSavepointCommand => Future.successful(SavepointResult(""))
     case DMTestScenarioCommand(_, canonicalProcess, scenarioTestData) =>
-      flinkTestRunner.test(canonicalProcess, scenarioTestData) // it's just for streaming e2e tests from file purposes
+      flinkTestRunner.runTestsAsync(
+        canonicalProcess,
+        scenarioTestData
+      ) // it's just for streaming e2e tests from file purposes
   }
 
   private def description(canonicalProcess: CanonicalProcess) = {

engine/development/deploymentManager/src/main/scala/pl/touk/nussknacker/development/manager/MockableDeploymentManagerProvider.scala

@@ -4,16 +4,17 @@ import cats.data.Validated.valid
 import cats.data.ValidatedNel
 import com.typesafe.config.Config
 import io.circe.Json
+import org.apache.flink.configuration.Configuration
 import pl.touk.nussknacker.development.manager.MockableDeploymentManagerProvider.MockableDeploymentManager
 import pl.touk.nussknacker.engine.ModelData.BaseModelDataExt
 import pl.touk.nussknacker.engine._
 import pl.touk.nussknacker.engine.api.component.ScenarioPropertyConfig
-import pl.touk.nussknacker.engine.api.definition.{NotBlankParameterValidator, StringParameterEditor}
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.deployment.simple.{SimpleProcessStateDefinitionManager, SimpleStateStatus}
 import pl.touk.nussknacker.engine.api.process.{ProcessIdWithName, ProcessName, VersionId}
 import pl.touk.nussknacker.engine.deployment.ExternalDeploymentId
-import pl.touk.nussknacker.engine.management.{FlinkProcessTestRunner, FlinkStreamingPropertiesConfig}
+import pl.touk.nussknacker.engine.management.FlinkStreamingPropertiesConfig
+import pl.touk.nussknacker.engine.management.testsmechanism.FlinkProcessTestRunner
 import pl.touk.nussknacker.engine.newdeployment.DeploymentId
 import pl.touk.nussknacker.engine.testing.StubbingCommands
 import pl.touk.nussknacker.engine.testmode.TestProcess.TestResults
@@ -59,7 +60,9 @@ object MockableDeploymentManagerProvider {
       with StubbingCommands {
 
     private lazy val testRunnerOpt =
-      modelDataOpt.map(modelData => new FlinkProcessTestRunner(modelData.asInvokableModelData))
+      modelDataOpt.map(modelData =>
+        new FlinkProcessTestRunner(modelData.asInvokableModelData, parallelism = 1, new Configuration())
+      )
 
     override def resolve(
         idWithName: ProcessIdWithName,
@@ -102,7 +105,7 @@ object MockableDeploymentManagerProvider {
             .get()
             .get(processVersion.processName.value)
             .map(Future.successful)
-            .orElse(testRunnerOpt.map(_.test(scenario, testData)))
+            .orElse(testRunnerOpt.map(_.runTestsAsync(scenario, testData)))
             .getOrElse(
               throw new IllegalArgumentException(
                 s"Tests results not mocked for scenario [${processVersion.processName.value}] and no model data provided"

engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/runner/FlinkProcessMain.scala

@@ -1,18 +1,21 @@
 package pl.touk.nussknacker.engine.process.runner
 
-import java.io.File
 import com.typesafe.config.{Config, ConfigFactory}
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.flink.api.common.ExecutionConfig
-import pl.touk.nussknacker.engine.{ModelConfigs, ModelData}
 import pl.touk.nussknacker.engine.api.{CirceUtil, ProcessVersion}
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
 import pl.touk.nussknacker.engine.deployment.DeploymentData
+import pl.touk.nussknacker.engine.marshall.ScenarioParser
 import pl.touk.nussknacker.engine.process.ExecutionConfigPreparer
+import pl.touk.nussknacker.engine.{ModelConfigs, ModelData}
 
+import java.io.File
+import java.nio.charset.StandardCharsets
+import scala.util.Using
 import scala.util.control.NonFatal
 
-trait FlinkProcessMain[Env] extends FlinkRunner with LazyLogging {
+trait FlinkProcessMain[Env] extends LazyLogging {
 
   def main(argsWithHack: Array[String]): Unit = {
     try {
@@ -61,6 +64,15 @@ trait FlinkProcessMain[Env] extends FlinkRunner with LazyLogging {
       prepareExecutionConfig: ExecutionConfigPreparer
   ): Unit
 
+  protected def readProcessFromArg(arg: String): CanonicalProcess = {
+    val canonicalJson = if (arg.startsWith("@")) {
+      Using.resource(scala.io.Source.fromFile(arg.substring(1), StandardCharsets.UTF_8.name()))(_.mkString)
+    } else {
+      arg
+    }
+    ScenarioParser.parseUnsafe(canonicalJson)
+  }
+
   private def parseProcessVersion(json: String): ProcessVersion =
     CirceUtil.decodeJsonUnsafe[ProcessVersion](json, "invalid scenario version")