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 e836fc9fa0 status on transform can change mid check, fixes #9096
(#9109)
e836fc9fa0 is described below
commit e836fc9fa05d6b8cd50e325b3b910048840dc25c
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Oct 30 10:23:41 2026 +0100
status on transform can change mid check, fixes #9096 (#9109)
---
.../BlockUntilTransformsFinish.java | 21 ++--
.../BlockUntilTransformsFinishTest.java | 116 +++++++++++++++++++++
2 files changed, 126 insertions(+), 11 deletions(-)
diff --git
a/plugins/transforms/blockuntiltransformsfinish/src/main/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinish.java
b/plugins/transforms/blockuntiltransformsfinish/src/main/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinish.java
index ac37ae185a..2a99a11f82 100644
---
a/plugins/transforms/blockuntiltransformsfinish/src/main/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinish.java
+++
b/plugins/transforms/blockuntiltransformsfinish/src/main/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinish.java
@@ -24,7 +24,7 @@ import org.apache.hop.core.exception.HopException;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
-import org.apache.hop.pipeline.engine.EngineComponent;
+import org.apache.hop.pipeline.engine.EngineComponent.ComponentExecutionStatus;
import org.apache.hop.pipeline.engine.IEngineComponent;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;
@@ -100,23 +100,22 @@ public class BlockUntilTransformsFinish
while (it.hasNext()) {
Entry<Integer, IEngineComponent> e = it.next();
IEngineComponent transform = e.getValue();
- if (transform.getStatus() ==
EngineComponent.ComponentExecutionStatus.STATUS_RUNNING
- || transform.getStatus() ==
EngineComponent.ComponentExecutionStatus.STATUS_IDLE
- || transform.getStatus() ==
EngineComponent.ComponentExecutionStatus.STATUS_INIT
- || transform.getStatus() ==
EngineComponent.ComponentExecutionStatus.STATUS_PAUSED) {
+ // Read the status once: it can change between reads, e.g. from idle
to running at
+ // startup, and then none of the checks below would match (#9096).
+ ComponentExecutionStatus status = transform.getStatus();
+ if (status == ComponentExecutionStatus.STATUS_RUNNING
+ || status == ComponentExecutionStatus.STATUS_IDLE
+ || status == ComponentExecutionStatus.STATUS_INIT
+ || status == ComponentExecutionStatus.STATUS_PAUSED) {
// This transform is still running...
data.continueLoop = true;
} else {
if (isBasic()) {
- logBasic(
- "Transform "
- + transform.getName()
- + " status: "
- + transform.getStatusDescription());
+ logBasic("Transform " + transform.getName() + " status: " +
status.getDescription());
}
// We have done with this transform.
// remove it from the map
- data.componentMap.remove(e.getKey());
+ it.remove();
if (isDetailed()) {
logDetailed(
"Finished running transform ["
diff --git
a/plugins/transforms/blockuntiltransformsfinish/src/test/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinishTest.java
b/plugins/transforms/blockuntiltransformsfinish/src/test/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinishTest.java
new file mode 100644
index 0000000000..6b7872a492
--- /dev/null
+++
b/plugins/transforms/blockuntiltransformsfinish/src/test/java/org/apache/hop/pipeline/transforms/blockuntiltransformsfinish/BlockUntilTransformsFinishTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.transforms.blockuntiltransformsfinish;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.pipeline.engine.EngineComponent.ComponentExecutionStatus;
+import org.apache.hop.pipeline.engine.IEngineComponent;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class BlockUntilTransformsFinishTest {
+ private static final String WAITED_FOR = "write to table";
+
+ private TransformMockHelper<BlockUntilTransformsFinishMeta,
BlockUntilTransformsFinishData>
+ helper;
+
+ @BeforeEach
+ void setup() {
+ helper =
+ new TransformMockHelper<>(
+ "BLOCK UNTIL TRANSFORMS FINISH TEST",
+ BlockUntilTransformsFinishMeta.class,
+ BlockUntilTransformsFinishData.class);
+ when(helper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(helper.iLogChannel);
+ }
+
+ @AfterEach
+ void tearDown() {
+ helper.cleanUp();
+ }
+
+ /**
+ * #9096: the waited-for transform switches from idle to running while its
status is checked. The
+ * first status read sees IDLE and every later read sees RUNNING, so a check
that reads the status
+ * more than once matches none of the "still busy" states and stops waiting.
+ */
+ @Test
+ void waitsWhenTransformStartsRunningDuringStatusCheck() throws Exception {
+ AtomicInteger statusReads = new AtomicInteger();
+ AtomicBoolean finished = new AtomicBoolean();
+ IEngineComponent waitedFor = mock(IEngineComponent.class);
+ when(waitedFor.getName()).thenReturn(WAITED_FOR);
+ when(waitedFor.getStatus())
+ .thenAnswer(
+ invocation -> {
+ int read = statusReads.incrementAndGet();
+ if (read == 1) {
+ return ComponentExecutionStatus.STATUS_IDLE;
+ }
+ if (read < 10) {
+ return ComponentExecutionStatus.STATUS_RUNNING;
+ }
+ finished.set(true);
+ return ComponentExecutionStatus.STATUS_FINISHED;
+ });
+ when(helper.pipeline.findComponent(WAITED_FOR, 0)).thenReturn(waitedFor);
+
+ BlockUntilTransformsFinishMeta meta = new BlockUntilTransformsFinishMeta();
+ meta.setBlockingTransforms(List.of(new BlockingTransform(WAITED_FOR,
"0")));
+
+ BlockUntilTransformsFinish transform =
+ spy(
+ new BlockUntilTransformsFinish(
+ helper.transformMeta,
+ meta,
+ new BlockUntilTransformsFinishData(),
+ 0,
+ helper.pipelineMeta,
+ helper.pipeline));
+
+ AtomicBoolean rowPassedBeforeFinish = new AtomicBoolean();
+ doReturn(new Object[] {1L}).when(transform).getRow();
+ doAnswer(
+ invocation -> {
+ rowPassedBeforeFinish.set(!finished.get());
+ return null;
+ })
+ .when(transform)
+ .putRow(any(), any());
+
+ transform.processRow();
+
+ assertFalse(rowPassedBeforeFinish.get(), "The row was passed on before the
transform finished");
+ assertTrue(finished.get());
+ }
+}
