Changes in sub workflow trigger decide on the parent asynchronously (#2125)

Sub workflow updates the parent's task and status, but triggers the decide async by pushing the parent's id to the decider queue.

Author: aravindanr

common/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java

@@ -199,6 +199,13 @@ public boolean isRetriable() {
     @ProtoField(id = 41)
     private String subWorkflowId;
 
+    /**
+     * Use to note that a sub workflow associated with SUB_WORKFLOW task
+     * has an action performed on it directly.
+     */
+    @ProtoField(id = 42)
+    private boolean subworkflowChanged;
+
     public Task() {
     }
 
@@ -735,6 +742,14 @@ public void setWorkflowPriority(int workflowPriority) {
         this.workflowPriority = workflowPriority;
     }
 
+    public boolean isSubworkflowChanged() {
+        return subworkflowChanged;
+    }
+
+    public void setSubworkflowChanged(boolean subworkflowChanged) {
+        this.subworkflowChanged = subworkflowChanged;
+    }
+
     public String getSubWorkflowId() {
         // For backwards compatibility
         if (StringUtils.isNotBlank(subWorkflowId)) {
@@ -787,6 +802,7 @@ public Task copy() {
         copy.setExecutionNameSpace(executionNameSpace);
         copy.setIsolationGroupId(isolationGroupId);
         copy.setSubWorkflowId(getSubWorkflowId());
+        copy.setSubworkflowChanged(subworkflowChanged);
 
         return copy;
     }
@@ -855,6 +871,7 @@ public String toString() {
             ", externalOutputPayloadStoragePath='" + externalOutputPayloadStoragePath + '\'' +
             ", isolationGroupId='" + isolationGroupId + '\'' +
             ", executionNameSpace='" + executionNameSpace + '\'' +
+            ", subworkflowChanged='" + subworkflowChanged + '\'' +
             '}';
     }
 

common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowDef.java

@@ -20,19 +20,19 @@
 import com.netflix.conductor.common.constraints.TaskReferenceNameUniqueConstraint;
 import com.netflix.conductor.common.metadata.Auditable;
 import com.netflix.conductor.common.metadata.tasks.TaskType;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+
 import javax.validation.Valid;
 import javax.validation.constraints.Email;
 import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 @ProtoMessage
 @TaskReferenceNameUniqueConstraint
@@ -303,6 +303,10 @@ public static String getKey(String name, int version) {
         return name + "." + version;
     }
 
+    public boolean containsType(String taskType) {
+        return collectTasks().stream().anyMatch(t -> t.getType().equals(taskType));
+    }
+
     public WorkflowTask getNextTask(String taskReferenceName) {
         WorkflowTask workflowTask = getTaskByRefName(taskReferenceName);
         if (workflowTask != null && TaskType.TERMINATE.name().equals(workflowTask.getType())) {

common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java

@@ -44,7 +44,7 @@ public class WorkflowTask {
     @Deprecated
     public enum Type {
         SIMPLE, DYNAMIC, FORK_JOIN, FORK_JOIN_DYNAMIC, DECISION, JOIN, SUB_WORKFLOW, EVENT, WAIT, USER_DEFINED;
-        private static Set<String> systemTasks = new HashSet<>();
+        private static final Set<String> systemTasks = new HashSet<>();
 
         static {
             systemTasks.add(Type.SIMPLE.name());

common/src/main/java/com/netflix/conductor/common/run/Workflow.java

@@ -1,14 +1,14 @@
 /*
- * Copyright 2020 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.
+ * Copyright 2021 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
+ *   <p>
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *   <p>
+ *   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 com.netflix.conductor.common.run;
 
@@ -18,6 +18,7 @@
 import com.netflix.conductor.common.metadata.Auditable;
 import com.netflix.conductor.common.metadata.tasks.Task;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
+import org.apache.commons.lang3.StringUtils;
 
 import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
@@ -485,6 +486,10 @@ public void setLastRetriedTime(long lastRetriedTime) {
         this.lastRetriedTime = lastRetriedTime;
     }
 
+    public boolean hasParent() {
+        return StringUtils.isNotEmpty(parentWorkflowId);
+    }
+
     public Task getTaskByRefName(String refName) {
         if (refName == null) {
             throw new RuntimeException(
@@ -541,6 +546,14 @@ public String toString() {
         return getWorkflowName() + "." + getWorkflowVersion() + "/" + workflowId + "." + status;
     }
 
+    /**
+     * A string representation of all relevant fields that identify this workflow.
+     * Intended for use in log and other system generated messages.
+     */
+    public String identifierString() {
+        return String.format("%s.%s/%s", getWorkflowName(), getWorkflowVersion(), workflowId);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

common/src/test/java/com/netflix/conductor/common/tasks/TaskTest.java

@@ -95,7 +95,7 @@ public void testTaskQueueWaitTime() {
     public void testDeepCopyTask() {
         final Task task = new Task();
         // In order to avoid forgetting putting inside the copy method the newly added fields check the number of declared fields.
-        final int expectedTaskFieldsNumber = 39;
+        final int expectedTaskFieldsNumber = 40;
         final int declaredFieldsNumber = task.getClass().getDeclaredFields().length;
 
         assertEquals(expectedTaskFieldsNumber, declaredFieldsNumber);
@@ -135,6 +135,7 @@ public void testDeepCopyTask() {
         task.setReasonForIncompletion("");
         task.setWorkerId("");
         task.setSubWorkflowId("");
+        task.setSubworkflowChanged(false);
 
         final Task copy = task.deepCopy();
         assertEquals(task, copy);

contribs/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java

@@ -29,6 +29,11 @@
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.GetResponse;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.Observable;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -43,10 +48,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import rx.Observable;
 
 /**
  * @author Ritu Parathody

contribs/src/test/java/com/netflix/conductor/contribs/metrics/PrometheusMetricsConfigurationTest.java

@@ -12,11 +12,8 @@
  */
 package com.netflix.conductor.contribs.metrics;
 
-import java.lang.reflect.Field;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
-
+import com.netflix.spectator.api.Meter;
+import com.netflix.spectator.api.Spectator;
 import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -25,8 +22,10 @@
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringRunner;
 
-import com.netflix.spectator.api.Meter;
-import com.netflix.spectator.api.Spectator;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
 
 @RunWith(SpringRunner.class)
 @TestPropertySource(properties = {"conductor.metrics-prometheus.enabled=true"})

contribs/src/test/java/com/netflix/conductor/contribs/queue/sqs/QueueManagerTest.java

@@ -18,9 +18,9 @@
 import com.netflix.conductor.common.metadata.tasks.Task;
 import com.netflix.conductor.common.metadata.tasks.Task.Status;
 import com.netflix.conductor.common.run.Workflow;
-import com.netflix.conductor.core.events.queue.QueueManager;
 import com.netflix.conductor.core.events.queue.Message;
 import com.netflix.conductor.core.events.queue.ObservableQueue;
+import com.netflix.conductor.core.events.queue.QueueManager;
 import com.netflix.conductor.core.execution.tasks.Wait;
 import com.netflix.conductor.service.ExecutionService;
 import org.junit.Before;

core/src/main/java/com/netflix/conductor/core/Lifecycle.java

@@ -1,30 +1,38 @@
 /*
- * Copyright 2021 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.
+ *
+ *  * Copyright 2021 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
+ *  * <p>
+ *  * http://www.apache.org/licenses/LICENSE-2.0
+ *  * <p>
+ *  * 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 com.netflix.conductor.core;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.context.SmartLifecycle;
 
-public class LifecycleAwareComponent implements SmartLifecycle {
+public abstract class LifecycleAwareComponent implements SmartLifecycle {
     private volatile boolean running = false;
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(LifecycleAwareComponent.class);
+
     @Override
     public void start() {
         running = true;
+        LOGGER.info("{} started.", getClass().getSimpleName());
     }
 
     @Override
     public void stop() {
         running = false;
+        LOGGER.info("{} stopped.", getClass().getSimpleName());
     }
 
     @Override