This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 63e6789e87 Transfroms aren't always correctly disposed, fixes #5970
(#7827)
63e6789e87 is described below
commit 63e6789e874f66f29d2939f9b693055c30b7da0a
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Aug 7 12:56:10 2026 +0200
Transfroms aren't always correctly disposed, fixes #5970 (#7827)
---
.../java/org/apache/hop/pipeline/Pipeline.java | 41 ++++---
.../engines/local/LocalPipelineEngine.java | 36 ++++--
.../engines/local/LocalPipelineEngineTest.java | 121 +++++++++++++++++++++
3 files changed, 177 insertions(+), 21 deletions(-)
diff --git a/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
b/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
index 01b383840c..d227687bef 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/Pipeline.java
@@ -1151,19 +1151,7 @@ public abstract class Pipeline
// Also explicitly call dispose() to clean up resources opened during
// init()
//
- for (TransformInitThread initThread : initThreads) {
- TransformMetaDataCombi combi = initThread.getCombi();
-
- // Dispose will overwrite the status, but we set it back right after
- // this.
- combi.transform.dispose();
-
- if (initThread.isOk()) {
- combi.data.setStatus(ComponentExecutionStatus.STATUS_HALTED);
- } else {
- combi.data.setStatus(ComponentExecutionStatus.STATUS_STOPPED);
- }
- }
+ disposeInitializedTransforms();
// Just for safety, fire the pipeline finished listeners...
try {
@@ -1198,6 +1186,33 @@ public abstract class Pipeline
setReadyToStart(true);
}
+ /**
+ * Dispose the transforms which were initialized in {@link
#prepareExecution()}. This releases the
+ * resources they acquired during init() (database connections, files, ...)
in the situations
+ * where the transform threads are never started and {@link RunThread} can't
do it for us.
+ */
+ public void disposeInitializedTransforms() {
+ if (transforms == null) {
+ return;
+ }
+ for (TransformMetaDataCombi combi : transforms) {
+ // Dispose will overwrite the status, but we set it back right after
this.
+ //
+ ComponentExecutionStatus status = combi.data.getStatus();
+ try {
+ combi.transform.dispose();
+ } catch (Exception e) {
+ // A transform which fails to clean up shouldn't keep the others from
doing so.
+ //
+ log.logError("Error disposing transform " + combi.transformName, e);
+ }
+ combi.data.setStatus(
+ status == ComponentExecutionStatus.STATUS_STOPPED
+ ? ComponentExecutionStatus.STATUS_STOPPED
+ : ComponentExecutionStatus.STATUS_HALTED);
+ }
+ }
+
/**
* Starts the threads prepared by prepareThreads(). Before you start the
threads, you can add
* RowListeners to them.
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
index c817fb3cab..5ce26d1d7e 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
@@ -235,17 +235,37 @@ public class LocalPipelineEngine extends Pipeline
implements IPipelineEngine<Pip
super.prepareExecution();
- // Do the lookup of the execution information only once
- lookupExecutionInformationLocation();
+ // The transforms are initialized at this point, which means they can hold
on to database
+ // connections, files, ... If anything below fails we never get to start
the transform threads,
+ // so nothing would ever dispose them. Clean up explicitly to avoid
leaking those resources.
+ //
+ try {
+ // Do the lookup of the execution information only once
+ lookupExecutionInformationLocation();
- // Register the pipeline
- registerPipelineExecutionInformation();
+ // Register the pipeline
+ registerPipelineExecutionInformation();
- // Attach samplers to the transforms if needed
- //
- addTransformExecutionSamplers();
+ // Attach samplers to the transforms if needed
+ //
+ addTransformExecutionSamplers();
+ } catch (Exception e) {
+ disposeInitializedTransforms();
- //
+ // Just for safety, fire the pipeline finished listeners...
+ //
+ try {
+ fireExecutionFinishedListeners();
+ } catch (HopException listenerException) {
+ log.logError(
+ "Error firing the execution finished listeners after a failed
preparation",
+ listenerException);
+ } finally {
+ // Flag the pipeline as finished even if an exception was thrown
+ setFinished(true);
+ }
+ throw e;
+ }
}
/**
diff --git
a/engine/src/test/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngineTest.java
b/engine/src/test/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngineTest.java
new file mode 100644
index 0000000000..94e6e77e18
--- /dev/null
+++
b/engine/src/test/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngineTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.engines.local;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engine.EngineComponent.ComponentExecutionStatus;
+import org.apache.hop.pipeline.transform.BaseTransformData;
+import org.apache.hop.pipeline.transform.ITransform;
+import org.apache.hop.pipeline.transform.ITransformData;
+import org.apache.hop.pipeline.transform.TransformMetaDataCombi;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(RestoreHopEngineEnvironmentExtension.class)
+class LocalPipelineEngineTest {
+
+ @BeforeAll
+ static void beforeClass() throws HopException {
+ HopEnvironment.init();
+ }
+
+ /**
+ * Transforms acquire resources (database connections, files, ...) in
init(). When the preparation
+ * of the pipeline fails after that init() the transform threads are never
started, so nothing
+ * else will ever release them. See issue #5970.
+ */
+ @Test
+ void transformsAreDisposedWhenPreparationFailsAfterInit() {
+ LocalPipelineEngine pipeline =
+ spy(
+ new LocalPipelineEngine(new PipelineMeta()) {
+ @Override
+ public void addTransformExecutionSamplers() throws HopException {
+ throw new HopException("Unable to attach the samplers");
+ }
+ });
+ pipeline.setLogChannel(mock(ILogChannel.class));
+
+ HopException exception = assertThrows(HopException.class,
pipeline::prepareExecution);
+
+ assertTrue(exception.getMessage().contains("Unable to attach the
samplers"));
+ verify(pipeline).disposeInitializedTransforms();
+ assertTrue(pipeline.isFinished(), "The pipeline should be flagged as
finished");
+ }
+
+ @Test
+ void disposeInitializedTransformsDisposesEveryTransform() throws Exception {
+ LocalPipelineEngine pipeline = new LocalPipelineEngine(new PipelineMeta());
+ pipeline.setLogChannel(mock(ILogChannel.class));
+ pipeline.prepareExecution();
+
+ // One transform which fails to clean up shouldn't keep the others from
doing so.
+ //
+ TransformMetaDataCombi failing =
+ combi("failing", ComponentExecutionStatus.STATUS_IDLE, new
RuntimeException("Oops"));
+ TransformMetaDataCombi initialized =
+ combi("initialized", ComponentExecutionStatus.STATUS_IDLE, null);
+ TransformMetaDataCombi notInitialized =
+ combi("not-initialized", ComponentExecutionStatus.STATUS_STOPPED,
null);
+ pipeline.getTransforms().add(failing);
+ pipeline.getTransforms().add(initialized);
+ pipeline.getTransforms().add(notInitialized);
+
+ pipeline.disposeInitializedTransforms();
+
+ verify(failing.transform).dispose();
+ verify(initialized.transform).dispose();
+ verify(notInitialized.transform).dispose();
+
+ assertEquals(ComponentExecutionStatus.STATUS_HALTED,
failing.data.getStatus());
+ assertEquals(ComponentExecutionStatus.STATUS_HALTED,
initialized.data.getStatus());
+ assertEquals(
+ ComponentExecutionStatus.STATUS_STOPPED,
+ notInitialized.data.getStatus(),
+ "A transform which never initialized should keep its stopped status");
+ }
+
+ private TransformMetaDataCombi combi(
+ String name, ComponentExecutionStatus status, RuntimeException
disposeError) {
+ ITransform transform = mock(ITransform.class);
+ if (disposeError != null) {
+ doThrow(disposeError).when(transform).dispose();
+ }
+ ITransformData data = new BaseTransformData() {};
+ data.setStatus(status);
+
+ TransformMetaDataCombi combi = new TransformMetaDataCombi();
+ combi.transformName = name;
+ combi.transform = transform;
+ combi.data = data;
+ return combi;
+ }
+}
