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 ef231f27a9 fix execute a process + tests, fixes #6990 (#7016)
ef231f27a9 is described below
commit ef231f27a9fa13928fcc79b06f4ccf04954c108c
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Apr 17 14:40:49 2026 +0200
fix execute a process + tests, fixes #6990 (#7016)
---
.../transforms/execprocess/ExecProcess.java | 114 +++++----
.../execprocess/ExecProcessMetaTest.java | 79 ++++++
.../transforms/execprocess/ExecProcessTest.java | 272 +++++++++++++++++++++
3 files changed, 423 insertions(+), 42 deletions(-)
diff --git
a/plugins/transforms/execprocess/src/main/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcess.java
b/plugins/transforms/execprocess/src/main/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcess.java
index 93f95d04cf..ef5574608e 100644
---
a/plugins/transforms/execprocess/src/main/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcess.java
+++
b/plugins/transforms/execprocess/src/main/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcess.java
@@ -24,9 +24,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
import org.apache.hop.core.exception.HopException;
@@ -224,59 +225,88 @@ public class ExecProcess extends
BaseTransform<ExecProcessMeta, ExecProcessData>
if (p == null) {
processresult.setErrorStream(errorMsg);
} else {
- CompletableFuture<IOException> future =
- p.onExit()
- .thenApply(
- processRef -> {
- try {
- // get output stream
- processresult.setOutputStream(
- getOutputString(
- new BufferedReader(
- new
InputStreamReader(processRef.getInputStream()))));
-
- // get error message
- processresult.setErrorStream(
- getOutputString(
- new BufferedReader(
- new
InputStreamReader(processRef.getErrorStream()))));
- } catch (IOException e) {
- return e;
- }
- return null;
- });
-
- // Wait until end
- IOException exception;
+ final Process child = p;
+ // Drain stdout/stderr while the process runs. Reading only after
onExit() can deadlock:
+ // if the child fills either pipe buffer before exiting, it blocks on
write forever.
+ CompletableFuture<String> stdoutFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return getOutputString(
+ new BufferedReader(new
InputStreamReader(child.getInputStream())));
+ } catch (IOException e) {
+ throw new CompletionException(e);
+ }
+ });
+ CompletableFuture<String> stderrFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return getOutputString(
+ new BufferedReader(new
InputStreamReader(child.getErrorStream())));
+ } catch (IOException e) {
+ throw new CompletionException(e);
+ }
+ });
+
while (true) {
try {
- exception = future.get(1, TimeUnit.SECONDS);
- break;
- } catch (TimeoutException ignore) {
- if (killing) {
- p.children().forEach(ProcessHandle::destroy);
- if (p.isAlive()) {
- p.destroy();
- }
- exception = future.get();
- logMinimal(BaseMessages.getString(PKG,
"ExecProcess.AbortProcess"));
+ if (child.waitFor(1, TimeUnit.SECONDS)) {
break;
}
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new HopException(
+ "Interrupted exception while running the process " +
Arrays.toString(process) + "!",
+ ie);
+ }
+ if (killing) {
+ child.children().forEach(ProcessHandle::destroy);
+ if (child.isAlive()) {
+ child.destroy();
+ }
+ try {
+ if (!child.waitFor(30, TimeUnit.SECONDS)) {
+ child.destroyForcibly();
+ child.waitFor();
+ }
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new HopException(
+ "Interrupted exception while running the process "
+ + Arrays.toString(process)
+ + "!",
+ ie);
+ }
+ logMinimal(BaseMessages.getString(PKG,
"ExecProcess.AbortProcess"));
+ break;
}
}
- if (exception != null) {
- throw exception;
+
+ try {
+ processresult.setOutputStream(stdoutFuture.get());
+ processresult.setErrorStream(stderrFuture.get());
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof IOException ioe) {
+ throw ioe;
+ }
+ if (cause instanceof RuntimeException re) {
+ throw re;
+ }
+ throw new IOException(cause);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new HopException(
+ "Interrupted exception while running the process " +
Arrays.toString(process) + "!",
+ ie);
}
- // get exit status
- processresult.setExistStatus(p.exitValue());
+ processresult.setExistStatus(child.exitValue());
}
} catch (IOException ioe) {
throw new HopException(
"IO exception while running the process " + Arrays.toString(process)
+ "!", ioe);
- } catch (InterruptedException ie) {
- throw new HopException(
- "Interrupted exception while running the process " +
Arrays.toString(process) + "!", ie);
} catch (Exception e) {
throw new HopException(e);
} finally {
diff --git
a/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessMetaTest.java
b/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessMetaTest.java
index 0652219cf8..13aa2e5eef 100644
---
a/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessMetaTest.java
+++
b/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessMetaTest.java
@@ -18,12 +18,28 @@
package org.apache.hop.pipeline.transforms.execprocess;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
class ExecProcessMetaTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
@Test
void testSerialization() throws Exception {
ExecProcessMeta meta =
@@ -39,4 +55,67 @@ class ExecProcessMetaTest {
assertEquals("value1", meta.getArgumentFields().get(0).getName());
assertEquals("value2", meta.getArgumentFields().get(1).getName());
}
+
+ @Test
+ void testCloneCopiesArgumentFieldsAndFlags() throws Exception {
+ HopEnvironment.init();
+
+ ExecProcessMeta a = new ExecProcessMeta();
+ a.setDefault();
+ a.setProcessField("p");
+ a.setArgumentsInFields(true);
+ ExecProcessMeta.EPField f = new ExecProcessMeta.EPField();
+ f.setName("arg0");
+ a.getArgumentFields().add(f);
+ a.setOutputLineDelimiter("|");
+
+ ExecProcessMeta b = a.clone();
+ assertNotSame(a, b);
+ assertEquals(a.getProcessField(), b.getProcessField());
+ assertTrue(b.isArgumentsInFields());
+ assertEquals(1, b.getArgumentFields().size());
+ assertEquals("arg0", b.getArgumentFields().get(0).getName());
+ assertEquals("|", b.getOutputLineDelimiter());
+ }
+
+ @Test
+ void testGetFieldsAppendsResultColumns() throws Exception {
+ HopEnvironment.init();
+
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setDefault();
+ meta.setResultFieldName("out");
+ meta.setErrorFieldName("err");
+ meta.setExitValueFieldName("exit");
+
+ IRowMeta row = new RowMeta();
+ row.addValueMeta(new ValueMetaString("in1"));
+
+ meta.getFields(
+ row,
+ "ExecProcess",
+ null,
+ new TransformMeta(),
+ new Variables(),
+ (IHopMetadataProvider) null);
+
+ assertEquals(4, row.size());
+ assertEquals("in1", row.getValueMeta(0).getName());
+ assertEquals("out", row.getValueMeta(1).getName());
+ assertEquals(ValueMetaString.TYPE_STRING, row.getValueMeta(1).getType());
+ assertEquals("err", row.getValueMeta(2).getName());
+ assertEquals("exit", row.getValueMeta(3).getName());
+ assertEquals(ValueMetaInteger.TYPE_INTEGER, row.getValueMeta(3).getType());
+ }
+
+ @Test
+ void testSupportsErrorHandlingMatchesFailWhenNotSuccess() throws Exception {
+ HopEnvironment.init();
+
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setFailWhenNotSuccess(false);
+ assertFalse(meta.supportsErrorHandling());
+ meta.setFailWhenNotSuccess(true);
+ assertTrue(meta.supportsErrorHandling());
+ }
}
diff --git
a/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessTest.java
b/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessTest.java
new file mode 100644
index 0000000000..a6a7938c40
--- /dev/null
+++
b/plugins/transforms/execprocess/src/test/java/org/apache/hop/pipeline/transforms/execprocess/ExecProcessTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.execprocess;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.QueueRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.transform.TransformMeta;
+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;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class ExecProcessTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ TransformMockHelper<ExecProcessMeta, ExecProcessData> smh;
+
+ @BeforeEach
+ void setUp() {
+ smh =
+ new TransformMockHelper<>(
+ "Execute a process", ExecProcessMeta.class, ExecProcessData.class);
+ when(smh.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(smh.iLogChannel);
+ when(smh.pipeline.isRunning()).thenReturn(true);
+ }
+
+ @AfterEach
+ void cleanUp() {
+ smh.cleanUp();
+ }
+
+ /**
+ * Executable + two args for {@link
ExecProcessMeta#setArgumentsInFields(boolean)} mode ({@code
+ * /bin/sh}, {@code -c}, script). Uses POSIX {@code /bin/sh} so the same
commands run on Linux and
+ * macOS; subprocess tests use {@link EnabledOnOs} for {@link OS#LINUX} and
{@link OS#MAC} only
+ * (not Windows).
+ */
+ private static String[] echoStdoutAndStderrCommand() {
+ return new String[] {"/bin/sh", "-c", "echo hop-out; echo hop-err 1>&2"};
+ }
+
+ /** Writes enough lines to stdout to exceed a typical OS pipe buffer
(regression for deadlock). */
+ private static String[] largeStdoutCommand() {
+ return new String[] {
+ "/bin/sh",
+ "-c",
+ "i=0; while [ $i -lt 2500 ]; do echo
AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA; i=$((i+1)); done",
+ };
+ }
+
+ private static String[] failingCommand() {
+ return new String[] {"/bin/sh", "-c", "exit 7"};
+ }
+
+ private static String singleTokenEchoCommand() {
+ return "/bin/echo hop-single";
+ }
+
+ private RowMeta inputRowMetaForArguments(String processField, String...
argFieldNames) {
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString(processField));
+ for (String name : argFieldNames) {
+ rowMeta.addValueMeta(new ValueMetaString(name));
+ }
+ return rowMeta;
+ }
+
+ private ExecProcessMeta metaForArgumentsMode(String processField, String...
argFieldNames) {
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setDefault();
+ meta.setProcessField(processField);
+ meta.setResultFieldName("result_out");
+ meta.setErrorFieldName("result_err");
+ meta.setExitValueFieldName("result_exit");
+ meta.setFailWhenNotSuccess(false);
+ meta.setArgumentsInFields(true);
+ for (String name : argFieldNames) {
+ ExecProcessMeta.EPField f = new ExecProcessMeta.EPField();
+ f.setName(name);
+ meta.getArgumentFields().add(f);
+ }
+ return meta;
+ }
+
+ private IRowSet inputRows(IRowMeta rowMeta, Object[]... rows) {
+ IRowSet rowSet = smh.getMockInputRowSet(rows);
+ when(rowSet.getRowMeta()).thenReturn(rowMeta);
+ return rowSet;
+ }
+
+ @Test
+ @EnabledOnOs({OS.LINUX, OS.MAC})
+ void processRow_argumentArray_capturesStdoutStderrAndExitCode() throws
HopException {
+ HopEnvironment.init();
+
+ String[] cmd = echoStdoutAndStderrCommand();
+ ExecProcessMeta meta = metaForArgumentsMode("cmd", "a1", "a2");
+ Object[] row = new Object[] {cmd[0], cmd[1], cmd[2]};
+ RowMeta rowMeta = inputRowMetaForArguments("cmd", "a1", "a2");
+
+ ExecProcess transform = newTransform(meta, rowMeta, row);
+
+ assertTrue(transform.processRow());
+ assertFalse(transform.processRow());
+
+ Object[] out = readSingleOutputRow(transform);
+ int base = rowMeta.size();
+ assertEquals("hop-out", out[base].toString().trim());
+ assertEquals("hop-err", out[base + 1].toString().trim());
+ assertEquals(0L, out[base + 2]);
+ }
+
+ @Test
+ @EnabledOnOs({OS.LINUX, OS.MAC})
+ void processRow_largeStdout_completesWithoutDeadlock() throws HopException {
+ HopEnvironment.init();
+
+ String[] cmd = largeStdoutCommand();
+ ExecProcessMeta meta = metaForArgumentsMode("cmd", "a1", "a2");
+ Object[] row = new Object[] {cmd[0], cmd[1], cmd[2]};
+ RowMeta rowMeta = inputRowMetaForArguments("cmd", "a1", "a2");
+
+ ExecProcess transform = newTransform(meta, rowMeta, row);
+
+ assertTrue(transform.processRow());
+ assertFalse(transform.processRow());
+
+ Object[] out = readSingleOutputRow(transform);
+ int base = rowMeta.size();
+ String stdout = (String) out[base];
+ assertNotNull(stdout);
+ assertTrue(stdout.length() > 50_000, "expected large stdout");
+ assertEquals(0L, out[base + 2]);
+ }
+
+ @Test
+ @EnabledOnOs({OS.LINUX, OS.MAC})
+ void processRow_singleCommandString_capturesOutput() throws HopException {
+ HopEnvironment.init();
+
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setDefault();
+ meta.setProcessField("cmdline");
+ meta.setResultFieldName("result_out");
+ meta.setErrorFieldName("result_err");
+ meta.setExitValueFieldName("result_exit");
+ meta.setArgumentsInFields(false);
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("cmdline"));
+
+ Object[] row = new Object[] {singleTokenEchoCommand()};
+ ExecProcess transform = newTransform(meta, rowMeta, row);
+
+ assertTrue(transform.processRow());
+ assertFalse(transform.processRow());
+
+ Object[] out = readSingleOutputRow(transform);
+ assertEquals("hop-single", out[1].toString().trim());
+ assertEquals(0L, out[3]);
+ }
+
+ @Test
+ @EnabledOnOs({OS.LINUX, OS.MAC})
+ void processRow_failWhenNotSuccess_setsErrors() throws HopException {
+ HopEnvironment.init();
+
+ String[] cmd = failingCommand();
+ ExecProcessMeta meta = metaForArgumentsMode("cmd", "a1", "a2");
+ Object[] row = new Object[] {cmd[0], cmd[1], cmd[2]};
+ RowMeta rowMeta = inputRowMetaForArguments("cmd", "a1", "a2");
+ meta.setFailWhenNotSuccess(true);
+
+ when(smh.transformMeta.isDoingErrorHandling()).thenReturn(false);
+
+ ExecProcess transform = newTransform(meta, rowMeta, row);
+
+ assertFalse(transform.processRow());
+ assertEquals(1, transform.getErrors());
+ }
+
+ @Test
+ void processRow_emptyProcessField_throws() throws HopException {
+ HopEnvironment.init();
+
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setDefault();
+ meta.setProcessField("cmdline");
+ meta.setResultFieldName("result_out");
+ meta.setErrorFieldName("result_err");
+ meta.setExitValueFieldName("result_exit");
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("cmdline"));
+
+ ExecProcess transform = newTransform(meta, rowMeta, new Object[] {""});
+
+ assertThrows(HopException.class, transform::processRow);
+ }
+
+ @Test
+ void init_requiresResultFieldName() throws HopException {
+ HopEnvironment.init();
+
+ ExecProcessMeta meta = new ExecProcessMeta();
+ meta.setDefault();
+ meta.setResultFieldName("");
+
+ ExecProcess transform =
+ new ExecProcess(
+ smh.transformMeta, meta, new ExecProcessData(), 0,
smh.pipelineMeta, smh.pipeline);
+
+ assertFalse(transform.init());
+ }
+
+ private ExecProcess newTransform(ExecProcessMeta meta, IRowMeta rowMeta,
Object[] row)
+ throws HopException {
+ TransformMeta tm = smh.transformMeta;
+ when(tm.isDoingErrorHandling()).thenReturn(false);
+
+ ExecProcess transform =
+ new ExecProcess(tm, meta, new ExecProcessData(), 0, smh.pipelineMeta,
smh.pipeline);
+ transform.init();
+ transform.setInputRowMeta(rowMeta);
+ transform.addRowSetToInputRowSets(inputRows(rowMeta, row));
+ transform.addRowSetToOutputRowSets(new QueueRowSet());
+ return transform;
+ }
+
+ private static Object[] readSingleOutputRow(ExecProcess transform) throws
HopException {
+ IRowSet outputRowSet = transform.getOutputRowSets().get(0);
+ Object[] out = outputRowSet.getRow();
+ assertNotNull(out);
+ return out;
+ }
+}
