[NU-1683] Fixes memory leak in test mechanism introduced in #4956

TouK · Jul 8, 2024 · 02037e7 · 02037e7
1 parent f809a64
commit 02037e7
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 2 deletions.
diff --git a/docs/Changelog.md b/docs/Changelog.md
@@ -21,6 +21,7 @@
 * [#6245](https://github.com/TouK/nussknacker/pull/6245) Parameter validations defined in AdditionalUIConfigProvider now properly impact dynamic components.
 * [#6264](https://github.com/TouK/nussknacker/pull/6264) Fix for DatabaseLookupEnricher mixing fields values when it is connected to ignite db
 * [#6270](https://github.com/TouK/nussknacker/pull/6270) Resolved an issue with comparing remote versions
+* [#6337](https://github.com/TouK/nussknacker/pull/6337) Fixes memory leak in test mechanism introduced in 1.13 version ([#4956](https://github.com/TouK/nussknacker/pull/4956))
 
 1.15.4 (5 July 2025)
 -------------------------

diff --git a/...ink/executor/src/main/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMain.scala b/...ink/executor/src/main/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMain.scala
@@ -46,16 +46,20 @@ class FlinkTestMain(
     val configuration: Configuration
 ) extends FlinkStubbedRunner {
 
-  def runTest: TestResults[Json] =
-    Using.resource(ResultsCollectingListenerHolder.registerTestEngineListener) { collectingListener =>
+  def runTest: TestResults[Json] = {
+    val collectingListener = ResultsCollectingListenerHolder.registerTestEngineListener
+    try {
       val resultCollector = new TestServiceInvocationCollector(collectingListener)
       val registrar       = prepareRegistrar(collectingListener, scenarioTestData)
       val env             = createEnv
 
       registrar.register(env, process, processVersion, deploymentData, resultCollector)
       execute(env, SavepointRestoreSettings.none())
       collectingListener.results
+    } finally {
+      collectingListener.clean()
     }
+  }
 
   protected def prepareRegistrar(
       collectingListener: ResultsCollectingListener[Json],

diff --git a/...mpiler/src/main/scala/pl/touk/nussknacker/engine/testmode/ResultsCollectingListener.scala b/...mpiler/src/main/scala/pl/touk/nussknacker/engine/testmode/ResultsCollectingListener.scala
@@ -24,6 +24,10 @@ case class ResultsCollectingListener[T](holderClass: String, runId: TestRunId, v
 
   def results: TestResults[T] = ResultsCollectingListenerHolder.resultsForId(runId)
 
+  // Warning! close can't clean resources because listener is passed into each scenario subpart and it will be closed few times
+  // We have to use dedicated clean() method when we are sure that we consumed results instead.
+  override final def close(): Unit = {}
+
   def clean(): Unit = ResultsCollectingListenerHolder.cleanResult(runId)
 
   override def nodeEntered(nodeId: String, context: Context, processMetaData: MetaData): Unit = {