add termination reason param to terminate task

.github/release-drafter.yml

@@ -35,3 +35,5 @@ version-resolver:
 
 exclude-labels:
   - 'skip-changelog'
+  - 'gradle-wrapper'
+  - 'github_actions'

core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java

@@ -805,9 +805,18 @@ private void endExecution(WorkflowModel workflow) {
                                     .getInputParameters()
                                     .get(Terminate.getTerminationStatusParameter());
             String reason =
-                    String.format(
-                            "Workflow is %s by TERMINATE task: %s",
-                            terminationStatus, terminateTask.get().getTaskId());
+                    (String)
+                            terminateTask
+                                    .get()
+                                    .getWorkflowTask()
+                                    .getInputParameters()
+                                    .get(Terminate.getTerminationReasonParameter());
+            if (StringUtils.isBlank(reason)) {
+                reason =
+                        String.format(
+                                "Workflow is %s by TERMINATE task: %s",
+                                terminationStatus, terminateTask.get().getTaskId());
+            }
             if (WorkflowModel.Status.FAILED.name().equals(terminationStatus)) {
                 workflow.setStatus(WorkflowModel.Status.FAILED);
                 workflow =
@@ -856,12 +865,6 @@ WorkflowModel completeWorkflow(WorkflowModel workflow) {
         deciderService.updateWorkflowOutput(workflow, null);
 
         workflow.setStatus(WorkflowModel.Status.COMPLETED);
-        workflow.setTasks(workflow.getTasks());
-        workflow.setOutput(workflow.getOutput());
-        workflow.setReasonForIncompletion(workflow.getReasonForIncompletion());
-        workflow.setFailedTaskId(workflow.getFailedTaskId());
-        workflow.setExternalOutputPayloadStoragePath(
-                workflow.getExternalOutputPayloadStoragePath());
 
         // update the failed reference task names
         workflow.getFailedReferenceTaskNames()

core/src/main/java/com/netflix/conductor/core/execution/tasks/Terminate.java

@@ -59,6 +59,7 @@
 public class Terminate extends WorkflowSystemTask {
 
     private static final String TERMINATION_STATUS_PARAMETER = "terminationStatus";
+    private static final String TERMINATION_REASON_PARAMETER = "terminationReason";
     private static final String TERMINATION_WORKFLOW_OUTPUT = "workflowOutput";
 
     public Terminate() {
@@ -84,6 +85,10 @@ public static String getTerminationStatusParameter() {
         return TERMINATION_STATUS_PARAMETER;
     }
 
+    public static String getTerminationReasonParameter() {
+        return TERMINATION_REASON_PARAMETER;
+    }
+
     public static String getTerminationWorkflowOutputParameter() {
         return TERMINATION_WORKFLOW_OUTPUT;
     }

docs/docs/configuration/systask.md

@@ -575,10 +575,11 @@ For example, if you have a decision where the first condition is met, you want t
 
 **Parameters:**
 
-|name|type|description|notes|
-|---|---|---|---|
-|terminationStatus|String|can only accept "COMPLETED" or "FAILED"|task cannot be optional|
-|workflowOutput|Any|Expected workflow output||
+|name|type| description                                       | notes                   |
+|---|---|---------------------------------------------------|-------------------------|
+|terminationStatus|String| can only accept "COMPLETED" or "FAILED"           | task cannot be optional |
+ |terminationReason|String| reason for incompletion to be set in the workflow | optional                |
+|workflowOutput|Any| Expected workflow output | optional |
 
 **Outputs:**
 
@@ -592,6 +593,7 @@ For example, if you have a decision where the first condition is met, you want t
   "taskReferenceName": "terminate0",
   "inputParameters": {
       "terminationStatus": "COMPLETED",
+	  "terminationReason": "",
       "workflowOutput": "${task0.output}"
   },
   "type": "TERMINATE",

test-harness/src/test/groovy/com/netflix/conductor/test/integration/FailureWorkflowSpec.groovy

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
  * <p>
  * Licensed 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
@@ -55,7 +55,7 @@ class FailureWorkflowSpec extends AbstractSpecification {
         with(workflowExecutionService.getExecutionStatus(workflowInstanceId, true)) {
             status == Workflow.WorkflowStatus.FAILED
             tasks.size() == 2
-            reasonForIncompletion.contains('Workflow is FAILED by TERMINATE task')
+            reasonForIncompletion == "Early exit in terminate"
             tasks[0].status == Task.Status.COMPLETED
             tasks[0].taskType == 'LAMBDA'
             tasks[0].seq == 1

test-harness/src/test/groovy/com/netflix/conductor/test/integration/LambdaAndTerminateTaskSpec.groovy

@@ -94,7 +94,7 @@ class LambdaAndTerminateTaskSpec extends AbstractSpecification {
         with(workflowExecutionService.getExecutionStatus(workflowInstanceId, true)) {
             status == Workflow.WorkflowStatus.FAILED
             tasks.size() == 2
-            reasonForIncompletion.contains('Workflow is FAILED by TERMINATE task')
+            reasonForIncompletion == "Early exit in terminate"
             tasks[0].status == Task.Status.COMPLETED
             tasks[0].taskType == 'LAMBDA'
             tasks[0].seq == 1

test-harness/src/test/resources/terminate_task_failed_workflow_integration.json

@@ -25,6 +25,7 @@
       "taskReferenceName": "terminate0",
       "inputParameters": {
         "terminationStatus": "FAILED",
+        "terminationReason": "Early exit in terminate",
         "workflowOutput": "${lambda0.output}"
       },
       "type": "TERMINATE",