[JBPM-9334] Process execution stuck at ExecWorkItemHandler node if co…

exec-workitem/src/main/java/org/jbpm/process/workitem/exec/ExecWorkItemHandler.java

@@ -16,12 +16,18 @@
 package org.jbpm.process.workitem.exec;
 
 import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.time.Period;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.exec.CommandLine;
 import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.ExecuteWatchdog;
 import org.apache.commons.exec.PumpStreamHandler;
 import org.jbpm.process.workitem.core.AbstractLogOrThrowWorkItemHandler;
 import org.jbpm.process.workitem.core.util.RequiredParameterValidator;
@@ -34,6 +40,8 @@
 import org.jbpm.process.workitem.core.util.service.WidService;
 import org.kie.api.runtime.process.WorkItem;
 import org.kie.api.runtime.process.WorkItemManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Wid(widfile = "ExecDefinitions.wid", name = "Exec",
         displayName = "Exec",
@@ -43,7 +51,8 @@
         icon = "Exec.png",
         parameters = {
                 @WidParameter(name = "Command", required = true),
-                @WidParameter(name = "Arguments", runtimeType = "java.util.List")
+                @WidParameter(name = "Arguments", runtimeType = "java.util.List"),
+                @WidParameter(name = "TimeoutInMillis", runtimeType = "java.lang.String")
         },
         results = {
                 @WidResult(name = "Output")
@@ -54,57 +63,109 @@
         serviceInfo = @WidService(category = "${name}", description = "${description}",
                 keywords = "execute,comand",
                 action = @WidAction(title = "Execute a command"),
-                authinfo = @WidAuth
-        ))
+                authinfo = @WidAuth))
 public class ExecWorkItemHandler extends AbstractLogOrThrowWorkItemHandler {
 
+    private static final Logger logger = LoggerFactory.getLogger(ExecWorkItemHandler.class);
     public static final String RESULT = "Output";
     private String parsedCommandStr = "";
+    private long defaultTimeout = 4000L;
 
     public void executeWorkItem(WorkItem workItem,
-                                WorkItemManager manager) {
+            WorkItemManager manager) {
 
         try {
 
             RequiredParameterValidator.validate(this.getClass(),
-                                                workItem);
+                    workItem);
 
             String command = (String) workItem.getParameter("Command");
             List<String> arguments = (List<String>) workItem.getParameter("Arguments");
+            String commandExecutionTimeout = (String) workItem.getParameter("TimeoutInMillis");
 
-            Map<String, Object> results = new HashMap<>();
-
-            CommandLine commandLine = CommandLine.parse(command);
-            if (arguments != null && arguments.size() > 0) {
-                commandLine.addArguments(arguments.toArray(new String[0]),
-                                         true);
+            if (commandExecutionTimeout != null) {
+                this.setDefaultTimeoutInSeconds(parsetimeout(commandExecutionTimeout));
             }
 
-            parsedCommandStr = commandLine.toString();
-
-            DefaultExecutor executor = new DefaultExecutor();
-            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
-            PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream);
-            executor.setStreamHandler(streamHandler);
-            executor.execute(commandLine);
+            String executionResult = executecommand(command, arguments, defaultTimeout);
 
+            Map<String, Object> results = new HashMap<>();
             results.put(RESULT,
-                        outputStream.toString());
-
-            outputStream.close();
+                    executionResult);
 
             manager.completeWorkItem(workItem.getId(),
-                                     results);
+                    results);
         } catch (Throwable t) {
             handleException(t);
         }
     }
 
+    protected long parsetimeout(String durationStr) {
+        try {
+            if (durationStr.startsWith("PT")) { // ISO-8601 PTnHnMn.nS
+                return Duration.parse(durationStr).toMillis();
+            } else if (!durationStr.contains("T")) { // ISO-8601 PnYnMnWnD
+                Period period = Period.parse(durationStr);
+                OffsetDateTime now = OffsetDateTime.now();
+                return Duration.between(now, now.plus(period)).toMillis();
+            } else { // ISO-8601 PnYnMnWnDTnHnMn.nS
+                String[] elements = durationStr.split("T");
+                Period period = Period.parse(elements[0]);
+                Duration duration = Duration.parse("PT" + elements[1]);
+                OffsetDateTime now = OffsetDateTime.now();
+
+                return Duration.between(now, now.plus(period).plus(duration)).toMillis();
+            }
+        } catch (Exception e) {
+            logger.error("Exception occured while parsing provided timeout" + durationStr
+                    + ".Default timeout of" + defaultTimeout
+                    + "ms will be used for command execution");
+            return defaultTimeout;
+        }
+    }
+
+    protected String executecommand(String command, List<String> arguments, long timeout) throws IOException {
+
+        String result = null;
+        CommandLine commandLine = CommandLine.parse(command);
+        if (arguments != null && arguments.size() > 0) {
+            commandLine.addArguments(arguments.toArray(new String[0]), true);
+        }
+
+        parsedCommandStr = commandLine.toString();
+
+        ExecuteWatchdog watchdog = new ExecuteWatchdog(timeout);
+        DefaultExecutor executor = new DefaultExecutor();
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream);
+        executor.setStreamHandler(streamHandler);
+        executor.setWatchdog(watchdog);
+        try {
+            executor.execute(commandLine);
+        } catch (IOException e) {
+            if (watchdog.killedProcess()) {
+                logger.error("A timeout occured after " + timeout
+                        + "ms while executing a command " + parsedCommandStr.replace(",", ""));
+                outputStream.reset();
+                outputStream.close();
+                return result = "A timeout occured after " + timeout
+                        + "ms while executing a command " + parsedCommandStr.replace(",", "");
+            }
+        }
+
+        return outputStream.toString();
+
+    }
+
     public void abortWorkItem(WorkItem workItem,
-                              WorkItemManager manager) {
+            WorkItemManager manager) {
         // Do nothing, this work item cannot be aborted
     }
 
+    public void setDefaultTimeoutInSeconds(long defaultTimeout) {
+        this.defaultTimeout = defaultTimeout;
+    }
+
     public String getParsedCommandStr() {
         return parsedCommandStr;
     }

exec-workitem/src/test/java/org/jbpm/process/workitem/exec/ExecWorkItemHandlerTest.java

@@ -85,6 +85,41 @@ public void testExecCommandWithArguments() throws Exception {
         assertNotNull(result);
         assertTrue(result.contains("java version") || result.contains("jdk version"));
     }
+
+	@Test(timeout = 6000)
+	public void testExecCommandWithTimeout() throws Exception {
+
+		TestWorkItemManager manager = new TestWorkItemManager();
+		WorkItemImpl workItem = new WorkItemImpl();
+		workItem.setParameter("Command",
+							  "ping");
+		List<String> argumentList = new ArrayList<>();
+		argumentList.add("127.0.0.1");
+		workItem.setParameter("Arguments",
+							argumentList);
+		workItem.setParameter("TimeoutInMillis", 
+								"PT5S");
+		ExecWorkItemHandler handler = new ExecWorkItemHandler();
+		handler.setLogThrownException(true);
+
+		handler.executeWorkItem(workItem,
+								manager);
+
+		assertNotNull(manager.getResults());
+		assertEquals(1,
+				manager.getResults().size());
+		assertTrue(manager.getResults().containsKey(workItem.getId()));
+
+		Map<String, Object> results = ((TestWorkItemManager) manager).getResults(workItem.getId());
+		String result = (String) results.get(ExecWorkItemHandler.RESULT);
+
+		assertEquals("[ping, 127.0.0.1]",
+				      handler.getParsedCommandStr());
+
+		assertNotNull(result);
+		assertTrue(result.contains("A timeout occured"));
+
+	}
 
     @Test
     public void testExecCommandInvalidParam() throws Exception {
@@ -99,6 +134,6 @@ public void testExecCommandInvalidParam() throws Exception {
 
         assertNotNull(manager.getResults());
         assertEquals(0,
-                     manager.getResults().size());
+                manager.getResults().size());
     }
 }