[NU-2045] FlinkMiniClusterWithServices: attached variants of methods creating StreamExecutionEnvironment + SyncIO to fix problem with ThreadLocal + clean up

Author: arkadius

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

@@ -79,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) = {
@@ -106,25 +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 = {
+    // 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)))
   }
 
+  // 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()
 
 }