[NU-1979] Added useMiniClusterForDeployment configuration option

.run/NussknackerApp-dist-config.run.xml

@@ -6,6 +6,7 @@
       <env name="AUTHENTICATION_USERS_FILE" value="../../../nussknacker-dist/src/universal/conf/users.conf" />
       <env name="CONSOLE_THRESHOLD_LEVEL" value="DEBUG" />
       <env name="FLINK_REST_URL" value="http://localhost:3031" />
+      <env name="FLINK_USE_MINICLUSTER_FOR_DEPLOY" value="true" />
       <env name="FLINK_SHOULD_VERIFY_BEFORE_DEPLOY" value="false" />
       <env name="GRAFANA_URL" value="http://localhost:8081/grafana" />
       <env name="INFLUXDB_URL" value="http://localhost:3086" />
@@ -18,6 +19,7 @@
       <env name="USAGE_REPORTS_SOURCE" value="sources" />
       <env name="INPUT_CONFIG_RESOLVE_ENV_VARIABLES" value="false" />
       <env name="TABLES_DEFINITION_FILE" value="../../../nussknacker-dist/src/universal/conf/dev-tables-definition.sql" />
+      <env name="WORKING_DIR" value="$PROJECT_DIR$/designer/server/work" />
     </envs>
     <option name="INCLUDE_PROVIDED_SCOPE" value="true" />
     <option name="MAIN_CLASS_NAME" value="pl.touk.nussknacker.ui.NussknackerApp" />

.run/NussknackerApp-postgres.run.xml

@@ -6,6 +6,7 @@
       <env name="AUTHENTICATION_USERS_FILE" value="../../../nussknacker-dist/src/universal/conf/users.conf" />
       <env name="CONSOLE_THRESHOLD_LEVEL" value="DEBUG" />
       <env name="FLINK_REST_URL" value="http://localhost:3031" />
+      <env name="FLINK_USE_MINICLUSTER_FOR_DEPLOY" value="true" />
       <env name="FLINK_SHOULD_VERIFY_BEFORE_DEPLOY" value="false" />
       <env name="GRAFANA_URL" value="http://localhost:8081/grafana" />
       <env name="INFLUXDB_URL" value="http://localhost:3086" />
@@ -22,7 +23,7 @@
       <env name="DB_USER" value="nu" />
       <env name="DB_PASSWORD" value="nupassword" />
       <env name="TABLES_DEFINITION_FILE" value="../../../nussknacker-dist/src/universal/conf/dev-tables-definition.sql" />
-      <env name="MANAGERS_DIR" value="managers" />
+      <env name="WORKING_DIR" value="$PROJECT_DIR$/designer/server/work" />
     </envs>
     <option name="INCLUDE_PROVIDED_SCOPE" value="true" />
     <option name="MAIN_CLASS_NAME" value="pl.touk.nussknacker.ui.NussknackerApp" />

.run/NussknackerApp.run.xml

@@ -1,24 +1,25 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="NussknackerApp" type="Application" factoryName="Application">
-    <option name="ALTERNATIVE_JRE_PATH" value="11" />
+    <option name="ALTERNATIVE_JRE_PATH" value="corretto-11" />
     <option name="ALTERNATIVE_JRE_PATH_ENABLED" value="true" />
     <envs>
       <env name="AUTHENTICATION_USERS_FILE" value="../../../nussknacker-dist/src/universal/conf/users.conf" />
       <env name="CONSOLE_THRESHOLD_LEVEL" value="DEBUG" />
       <env name="FLINK_REST_URL" value="http://localhost:3031" />
+      <env name="FLINK_USE_MINICLUSTER_FOR_DEPLOY" value="true" />
       <env name="FLINK_SHOULD_VERIFY_BEFORE_DEPLOY" value="false" />
       <env name="GRAFANA_URL" value="http://localhost:8081/grafana" />
       <env name="INFLUXDB_URL" value="http://localhost:3086" />
       <env name="INPUT_CONFIG_RESOLVE_ENV_VARIABLES" value="false" />
       <env name="KAFKA_ADDRESS" value="localhost:3032" />
-      <env name="MANAGERS_DIR" value="managers" />
       <env name="NUSSKNACKER_LOG_LEVEL" value="DEBUG" />
       <env name="OPENAPI_SERVICE_URL" value="http://localhost:5000" />
       <env name="SCHEMA_REGISTRY_URL" value="http://localhost:3082" />
       <env name="SQL_ENRICHER_URL" value="localhost:5432" />
       <env name="TABLES_DEFINITION_FILE" value="../../../nussknacker-dist/src/universal/conf/dev-tables-definition.sql" />
       <env name="USAGE_REPORTS_FINGERPRINT" value="development" />
       <env name="USAGE_REPORTS_SOURCE" value="sources" />
+      <env name="WORKING_DIR" value="$PROJECT_DIR$/designer/server/work" />
     </envs>
     <option name="INCLUDE_PROVIDED_SCOPE" value="true" />
     <option name="MAIN_CLASS_NAME" value="pl.touk.nussknacker.ui.NussknackerApp" />

designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala

@@ -141,11 +141,7 @@ sealed trait SchedulingSupport
 
 trait SchedulingSupported extends SchedulingSupport {
 
-  def createScheduledExecutionPerformer(
-      modelData: BaseModelData,
-      dependencies: DeploymentManagerDependencies,
-      deploymentConfig: Config,
-  ): ScheduledExecutionPerformer
+  def createScheduledExecutionPerformer(deploymentConfig: Config): ScheduledExecutionPerformer
 
   def customSchedulePropertyExtractorFactory: Option[SchedulePropertyExtractorFactory] = None
 

designer/server/src/main/resources/defaultDesignerConfig.conf

@@ -3,7 +3,6 @@
 # models and external tools (grafana, flink etc.). All models configurations also shouldn't be in this file
 
 managersDirs: ["./managers"]
-managersDirs: [ ${?MANAGERS_DIR} ]
 
 storageDir: ./storage
 storageDir: ${?STORAGE_DIR}

designer/server/src/main/scala/pl/touk/nussknacker/ui/process/periodic/PeriodicDeploymentManagerDecorator.scala

@@ -93,8 +93,7 @@ object PeriodicDeploymentManagerDecorator extends LazyLogging {
       delegate = underlying,
       dependencies = dependencies,
       periodicProcessesRepository = periodicProcessesRepository,
-      scheduledExecutionPerformer =
-        schedulingSupported.createScheduledExecutionPerformer(modelData, dependencies, deploymentConfig),
+      scheduledExecutionPerformer = schedulingSupported.createScheduledExecutionPerformer(deploymentConfig),
       schedulePropertyExtractorFactory = schedulePropertyExtractorFactory,
       processConfigEnricherFactory = processConfigEnricherFactory,
       listenerFactory = periodicProcessListenerFactory,

designer/server/src/main/scala/pl/touk/nussknacker/ui/server/AkkaHttpBasedRouteProvider.scala

@@ -809,7 +809,6 @@ class AkkaHttpBasedRouteProvider(
       additionalConfigsFromProvider,
       DesignerWideComponentId.default(processingType, _),
       workingDirectoryOpt = None, // we use the default working directory
-      _ => true,
       componentDefinitionExtractionMode,
     )
   }

designer/server/src/test/scala/pl/touk/nussknacker/test/config/WithFlinkContainersDeploymentManager.scala

@@ -1,9 +1,9 @@
 package pl.touk.nussknacker.test.config
 
-import com.dimafeng.testcontainers.{Container, ForAllTestContainer, LazyContainer, MultipleContainers}
+import com.dimafeng.testcontainers.{Container, ForAllTestContainer, MultipleContainers}
 import com.typesafe.config.Config
 import com.typesafe.config.ConfigValueFactory.fromAnyRef
-import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.scalalogging.StrictLogging
 import org.apache.commons.io.FileUtils
 import org.scalatest.{BeforeAndAfterAll, Suite}
 import pl.touk.nussknacker.engine.flink.test.docker.WithFlinkContainers
@@ -15,7 +15,7 @@ trait WithFlinkContainersDeploymentManager
     with ForAllTestContainer
     with WithFlinkContainers
     with BeforeAndAfterAll {
-  self: Suite with LazyLogging =>
+  self: Suite with StrictLogging =>
 
   override val container: Container = MultipleContainers(flinkContainers: _*)
 

designer/server/src/test/scala/pl/touk/nussknacker/test/mock/MockDeploymentManager.scala

@@ -5,14 +5,17 @@ import cats.data.Validated.valid
 import cats.data.ValidatedNel
 import cats.effect.unsafe.IORuntime
 import com.typesafe.config.Config
+import org.apache.flink.configuration.Configuration
 import sttp.client3.testing.SttpBackendStub
 import pl.touk.nussknacker.engine._
 import pl.touk.nussknacker.engine.api.ProcessVersion
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus
 import pl.touk.nussknacker.engine.api.process.ProcessName
 import pl.touk.nussknacker.engine.deployment._
+import pl.touk.nussknacker.engine.flink.minicluster.FlinkMiniClusterFactory
 import pl.touk.nussknacker.engine.flink.minicluster.scenariotesting.ScenarioStateVerificationConfig
+import pl.touk.nussknacker.engine.management.jobrunner.FlinkScenarioJobRunner
 import pl.touk.nussknacker.engine.management.{FlinkConfig, FlinkDeploymentManager, FlinkDeploymentManagerProvider}
 import pl.touk.nussknacker.engine.util.loader.{DeploymentManagersClassLoader, ModelClassLoader}
 import pl.touk.nussknacker.test.config.ConfigWithScalaVersion
@@ -43,7 +46,12 @@ class MockDeploymentManager private (
       modelData,
       deploymentManagerDependencies,
       FlinkConfig(None, scenarioStateVerification = ScenarioStateVerificationConfig(enabled = false)),
-      FlinkClientStub
+      Some(
+        FlinkMiniClusterFactory
+          .createMiniClusterWithServices(modelData.modelClassLoader, new Configuration, new Configuration)
+      ),
+      FlinkClientStub,
+      FlinkScenarioJobRunnerStub
     ) {
 
   import deploymentManagerDependencies._
@@ -105,6 +113,17 @@ class MockDeploymentManager private (
 
 }
 
+// This stub won't be used because we override the whole runDeployment method
+object FlinkScenarioJobRunnerStub extends FlinkScenarioJobRunner {
+
+  override def runScenarioJob(
+      command: DMRunDeploymentCommand,
+      savepointPathOpt: Option[String]
+  ): Future[Option[ExternalDeploymentId]] =
+    Future.failed(new IllegalAccessException("This implementation shouldn't be used"))
+
+}
+
 object MockDeploymentManager {
 
   def create(

designer/server/src/test/scala/pl/touk/nussknacker/test/mock/StubModelDataWithModelDefinition.scala

@@ -23,8 +23,7 @@ class StubModelDataWithModelDefinition(
 
   override def modelClassLoader: ModelClassLoader = ModelClassLoader.empty
 
-  override def modelConfigLoader: ModelConfigLoader =
-    new DefaultModelConfigLoader(_ => true)
+  override def modelConfigLoader: ModelConfigLoader = DefaultModelConfigLoader
 
   override def namingStrategy: NamingStrategy = NamingStrategy.Disabled
 

designer/server/src/test/scala/pl/touk/nussknacker/test/utils/domain/TestFactory.scala

@@ -38,7 +38,7 @@ import pl.touk.nussknacker.ui.process.processingtype.{
   ValueWithRestriction
 }
 import pl.touk.nussknacker.ui.process.repository._
-import pl.touk.nussknacker.ui.process.repository.activities.{DbScenarioActivityRepository, ScenarioActivityRepository}
+import pl.touk.nussknacker.ui.process.repository.activities.DbScenarioActivityRepository
 import pl.touk.nussknacker.ui.process.version.{ScenarioGraphVersionRepository, ScenarioGraphVersionService}
 import pl.touk.nussknacker.ui.security.api.{LoggedUser, RealLoggedUser}
 import pl.touk.nussknacker.ui.uiresolving.UIProcessResolver
@@ -129,7 +129,6 @@ object TestFactory {
       TestAdditionalUIConfigProvider.componentAdditionalConfigMap,
       componentId => DesignerWideComponentId(componentId.toString),
       workingDirectoryOpt = None,
-      _ => true,
       ComponentDefinitionExtractionMode.FinalAndBasicDefinitions
     )
 

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/BaseDeploymentApiHttpServiceBusinessSpec.scala

@@ -1,15 +1,15 @@
 package pl.touk.nussknacker.ui.api
 
-import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.scalalogging.StrictLogging
 import io.restassured.RestAssured.`given`
 import io.restassured.module.scala.RestAssuredSupport.AddThenToResponse
 import org.apache.commons.io.FileUtils
 import org.hamcrest.Matchers.{anyOf, equalTo}
-import org.scalatest.{LoneElement, Suite}
 import org.scalatest.concurrent.Eventually
 import org.scalatest.concurrent.PatienceConfiguration.{Interval, Timeout}
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.time.{Seconds, Span}
+import org.scalatest.{LoneElement, Suite}
 import org.testcontainers.containers.BindMode
 import pl.touk.nussknacker.engine.api.deployment.DeploymentStatusName
 import pl.touk.nussknacker.engine.build.ScenarioBuilder
@@ -30,7 +30,7 @@ import java.nio.file.{Files, Path}
 trait BaseDeploymentApiHttpServiceBusinessSpec extends WithFlinkContainersDeploymentManager {
   self: NuItTest
     with Suite
-    with LazyLogging
+    with StrictLogging
     with Matchers
     with Eventually
     with LoneElement

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpServiceBusinessSpec.scala

@@ -1,7 +1,7 @@
 package pl.touk.nussknacker.ui.api
 
 import cats.implicits.toTraverseOps
-import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.scalalogging.StrictLogging
 import io.restassured.RestAssured.`given`
 import io.restassured.module.scala.RestAssuredSupport.AddThenToResponse
 import org.apache.commons.io.FileUtils
@@ -34,7 +34,7 @@ class DeploymentApiHttpServiceBusinessSpec
     with WithBusinessCaseRestAssuredUsersExtensions
     with NuRestAssureMatchers
     with RestAssuredVerboseLoggingIfValidationFails
-    with LazyLogging
+    with StrictLogging
     with VeryPatientScalaFutures
     with Matchers
     with LoneElement {

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpServiceDeploymentCommentSpec.scala

@@ -1,7 +1,7 @@
 package pl.touk.nussknacker.ui.api
 
 import com.typesafe.config.{Config, ConfigValueFactory}
-import com.typesafe.scalalogging.LazyLogging
+import com.typesafe.scalalogging.StrictLogging
 import io.restassured.RestAssured.`given`
 import io.restassured.module.scala.RestAssuredSupport.AddThenToResponse
 import org.apache.commons.io.FileUtils
@@ -31,7 +31,7 @@ class DeploymentApiHttpServiceDeploymentCommentSpec
     with WithBusinessCaseRestAssuredUsersExtensions
     with NuRestAssureMatchers
     with RestAssuredVerboseLoggingIfValidationFails
-    with LazyLogging
+    with StrictLogging
     with VeryPatientScalaFutures
     with Matchers
     with LoneElement {

designer/server/src/test/scala/pl/touk/nussknacker/ui/process/periodic/flink/ScheduledExecutionPerformerTest.scala

@@ -8,7 +8,8 @@ import pl.touk.nussknacker.engine.api.ProcessVersion
 import pl.touk.nussknacker.engine.api.deployment.scheduler.model.{DeploymentWithRuntimeParams, RuntimeParams}
 import pl.touk.nussknacker.engine.api.deployment.scheduler.services.ScheduledExecutionPerformer
 import pl.touk.nussknacker.engine.api.process.{ProcessName, VersionId}
-import pl.touk.nussknacker.engine.management.{FlinkModelJarProvider, FlinkScheduledExecutionPerformer}
+import pl.touk.nussknacker.engine.management.FlinkScheduledExecutionPerformer
+import pl.touk.nussknacker.engine.management.jobrunner.FlinkModelJarProvider
 import pl.touk.nussknacker.engine.modelconfig.InputConfigDuringExecution
 import pl.touk.nussknacker.test.PatientScalaFutures