Merge branch 'minicluster-SyncIO-attached' into flink-sql-validations-using-env

Author: mslabek

engine/flink/minicluster/src/main/scala/pl/touk/nussknacker/engine/flink/minicluster/FlinkMiniClusterFactory.scala

@@ -1,8 +1,8 @@
 package pl.touk.nussknacker.engine.flink.minicluster
 
-import cats.effect.IO
+import cats.Applicative
+import cats.effect.{Sync, SyncIO}
 import cats.effect.kernel.Resource
-import cats.effect.unsafe.implicits.global
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.flink.configuration._
 import org.apache.flink.core.fs.FileSystem
@@ -16,6 +16,7 @@ import pl.touk.nussknacker.engine.util.ThreadUtils
 import pl.touk.nussknacker.engine.util.loader.ModelClassLoader
 
 import java.net.URLClassLoader
+import scala.language.higherKinds
 
 object FlinkMiniClusterFactory extends LazyLogging {
 
@@ -78,15 +79,7 @@ object FlinkMiniClusterFactory extends LazyLogging {
       miniCluster.start()
     }
 
-    def createStreamExecutionEnv(attached: Boolean): StreamExecutionEnvironment = {
-      FlinkMiniClusterStreamExecutionEnvironmentFactory.createStreamExecutionEnvironment(
-        miniCluster,
-        modelClassLoader,
-        attached
-      )
-    }
-
-    new FlinkMiniClusterWithServices(miniCluster, createStreamExecutionEnv)
+    new FlinkMiniClusterWithServices(miniCluster, modelClassLoader)
   }
 
   private def createMiniCluster(configuration: Configuration) = {
@@ -105,17 +98,40 @@ object FlinkMiniClusterFactory extends LazyLogging {
 
 class FlinkMiniClusterWithServices(
     val miniCluster: MiniCluster,
-    streamExecutionEnvironmentFactory: Boolean => StreamExecutionEnvironment
+    modelClassLoader: URLClassLoader
 ) extends AutoCloseable {
 
   def withDetachedStreamExecutionEnvironment[T](action: StreamExecutionEnvironment => T): T = {
-    createDetachedStreamExecutionEnvironment.use(env => IO(action(env))).unsafeRunSync()
+    // We use SyncIO, because passed actions sometimes uses ThreadLocal and we don't want to change the Thread which run this action
+    createDetachedStreamExecutionEnvironment[SyncIO].use(env => SyncIO.pure(action(env))).unsafeRunSync()
+  }
+
+  def createDetachedStreamExecutionEnvironment[F[_]: Sync]: Resource[F, StreamExecutionEnvironment] = {
+    Resource.fromAutoCloseable(Applicative[F].pure(streamExecutionEnvironmentFactory(false)))
   }
 
-  def createDetachedStreamExecutionEnvironment: Resource[IO, StreamExecutionEnvironment] = {
-    Resource.fromAutoCloseable(IO(streamExecutionEnvironmentFactory(false)))
+  // This method is used only by external project. It should be used with caution because action on StreamExecutionEnvironment
+  // can be blocking and it can cause thread pool starvation or deadlock. As an alternative, we recommend to use withDetachedStreamExecutionEnvironment
+  // combined with MiniClusterJobStatusCheckingOps
+  def withAttachedStreamExecutionEnvironment[T](action: StreamExecutionEnvironment => T): T = {
+    // We use SyncIO, because passed actions sometimes uses ThreadLocal and we don't want to change the Thread which run this action
+    createAttachedStreamExecutionEnvironment[SyncIO].use(env => SyncIO.pure(action(env))).unsafeRunSync()
   }
 
+  // This method is used only by external project. It should be used with caution because action on StreamExecutionEnvironment
+  // can be blocking and it can cause thread pool starvation or deadlock. As an alternative, we recommend to use createDetachedStreamExecutionEnvironment
+  // combined with MiniClusterJobStatusCheckingOps
+  def createAttachedStreamExecutionEnvironment[F[_]: Sync]: Resource[F, StreamExecutionEnvironment] = {
+    Resource.fromAutoCloseable(Applicative[F].pure(streamExecutionEnvironmentFactory(true)))
+  }
+
+  private def streamExecutionEnvironmentFactory(attached: Boolean): StreamExecutionEnvironment =
+    FlinkMiniClusterStreamExecutionEnvironmentFactory.createStreamExecutionEnvironment(
+      miniCluster,
+      modelClassLoader,
+      attached
+    )
+
   override def close(): Unit = miniCluster.close()
 
 }

engine/flink/minicluster/src/main/scala/pl/touk/nussknacker/engine/flink/minicluster/scenariotesting/FlinkMiniClusterScenarioStateVerifier.scala

@@ -57,7 +57,8 @@ class FlinkMiniClusterScenarioStateVerifier(
             env
           )
         val scenarioName = processVersion.processName
-        miniClusterWithServices.createDetachedStreamExecutionEnvironment
+        miniClusterWithServices
+          .createDetachedStreamExecutionEnvironment[IO]
           .use { env =>
             logger.info(s"Starting to verify $scenarioName")
             (for {

engine/flink/minicluster/src/main/scala/pl/touk/nussknacker/engine/flink/minicluster/scenariotesting/FlinkMiniClusterScenarioTestRunner.scala

@@ -60,7 +60,7 @@ class FlinkMiniClusterScenarioTestRunner(
         )
       }
       (for {
-        env                <- miniClusterWithServices.createDetachedStreamExecutionEnvironment
+        env                <- miniClusterWithServices.createDetachedStreamExecutionEnvironment[IO]
         collectingListener <- ResultsCollectingListenerHolder.registerTestEngineListener
       } yield (env, collectingListener))
         .use { case (env, collectingListener) =>