* 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 diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/config/CassandraProperties.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/config/CassandraProperties.java index 19286cad45..28d3eee972 100644 --- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/config/CassandraProperties.java +++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/config/CassandraProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 Netflix, Inc. + * Copyright 2022 Netflix, Inc. *
* 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 diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraBaseDAO.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraBaseDAO.java index 70664c1694..c576be8e91 100644 --- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraBaseDAO.java +++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraBaseDAO.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 Netflix, Inc. + * Copyright 2022 Netflix, Inc. *
* 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 diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAO.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAO.java index ea0b12de9f..ce797cea9e 100644 --- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAO.java +++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAO.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 Netflix, Inc. + * Copyright 2022 Netflix, Inc. *
* 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 diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java index 74e2016da8..4341186d70 100644 --- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java +++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 Netflix, Inc. + * Copyright 2022 Netflix, Inc. *
* 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
@@ -27,15 +27,15 @@
import com.netflix.conductor.cassandra.config.CassandraProperties;
import com.netflix.conductor.cassandra.util.Statements;
import com.netflix.conductor.common.metadata.events.EventExecution;
-import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskDef;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.common.utils.RetryUtil;
import com.netflix.conductor.core.exception.ApplicationException;
import com.netflix.conductor.core.exception.ApplicationException.Code;
import com.netflix.conductor.dao.ConcurrentExecutionLimitDAO;
import com.netflix.conductor.dao.ExecutionDAO;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.datastax.driver.core.BatchStatement;
import com.datastax.driver.core.PreparedStatement;
@@ -56,7 +56,6 @@
import static com.netflix.conductor.cassandra.util.Constants.TOTAL_PARTITIONS_KEY;
import static com.netflix.conductor.cassandra.util.Constants.TOTAL_TASKS_KEY;
import static com.netflix.conductor.cassandra.util.Constants.WORKFLOW_ID_KEY;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.IN_PROGRESS;
@Trace
public class CassandraExecutionDAO extends CassandraBaseDAO
@@ -172,11 +171,11 @@ public CassandraExecutionDAO(
}
@Override
- public List
* 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
diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Constants.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Constants.java
index f5eb9f7dfe..473c23132c 100644
--- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Constants.java
+++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Constants.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Statements.java b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Statements.java
index 5c538c41e7..a6ea122538 100644
--- a/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Statements.java
+++ b/cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/util/Statements.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAOSpec.groovy b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAOSpec.groovy
index 912d36c65c..214f3722de 100644
--- a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAOSpec.groovy
+++ b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraEventHandlerDAOSpec.groovy
@@ -1,20 +1,20 @@
/*
- * Copyright 2021 Netflix, Inc.
- *
- * 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
- *
- * 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.
+ * Copyright 2022 Netflix, Inc.
+ *
+ * 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
+ *
+ * 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 com.netflix.conductor.cassandra.dao
import com.netflix.conductor.common.metadata.events.EventExecution
import com.netflix.conductor.common.metadata.events.EventHandler
+
import spock.lang.Subject
class CassandraEventHandlerDAOSpec extends CassandraSpec {
diff --git a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraExecutionDAOSpec.groovy b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraExecutionDAOSpec.groovy
index 11727500ae..2a3b81c390 100644
--- a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraExecutionDAOSpec.groovy
+++ b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraExecutionDAOSpec.groovy
@@ -1,26 +1,26 @@
/*
- * Copyright 2021 Netflix, Inc.
- *
- * 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
- *
- * 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.
+ * Copyright 2022 Netflix, Inc.
+ *
+ * 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
+ *
+ * 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 com.netflix.conductor.cassandra.dao
import com.netflix.conductor.common.metadata.events.EventExecution
-import com.netflix.conductor.common.metadata.tasks.Task
import com.netflix.conductor.common.metadata.tasks.TaskDef
import com.netflix.conductor.common.metadata.workflow.WorkflowDef
import com.netflix.conductor.common.metadata.workflow.WorkflowTask
-import com.netflix.conductor.common.run.Workflow
import com.netflix.conductor.core.exception.ApplicationException
import com.netflix.conductor.core.utils.IDGenerator
+import com.netflix.conductor.model.TaskModel
+import com.netflix.conductor.model.WorkflowModel
+
import spock.lang.Subject
import static com.netflix.conductor.common.metadata.events.EventExecution.Status.COMPLETED
@@ -40,8 +40,8 @@ class CassandraExecutionDAOSpec extends CassandraSpec {
def tasks = []
// create tasks for a workflow and add to list
- Task task1 = new Task(workflowInstanceId: 'uuid', taskId: 'task1id', referenceTaskName: 'task1')
- Task task2 = new Task(workflowInstanceId: 'uuid', taskId: 'task2id', referenceTaskName: 'task2')
+ TaskModel task1 = new TaskModel(workflowInstanceId: 'uuid', taskId: 'task1id', referenceTaskName: 'task1')
+ TaskModel task2 = new TaskModel(workflowInstanceId: 'uuid', taskId: 'task2id', referenceTaskName: 'task2')
tasks << task1 << task2
when:
@@ -52,7 +52,7 @@ class CassandraExecutionDAOSpec extends CassandraSpec {
and:
// add a task from a different workflow to the list
- Task task3 = new Task(workflowInstanceId: 'other-uuid', taskId: 'task3id', referenceTaskName: 'task3')
+ TaskModel task3 = new TaskModel(workflowInstanceId: 'other-uuid', taskId: 'task3id', referenceTaskName: 'task3')
tasks << task3
when:
@@ -69,11 +69,11 @@ class CassandraExecutionDAOSpec extends CassandraSpec {
WorkflowDef workflowDef = new WorkflowDef()
workflowDef.name = "def1"
workflowDef.setVersion(1)
- Workflow workflow = new Workflow()
+ WorkflowModel workflow = new WorkflowModel()
workflow.setWorkflowDefinition(workflowDef)
workflow.setWorkflowId(workflowId)
workflow.setInput(new HashMap<>())
- workflow.setStatus(Workflow.WorkflowStatus.RUNNING)
+ workflow.setStatus(WorkflowModel.Status.RUNNING)
workflow.setCreateTime(System.currentTimeMillis())
when:
@@ -85,14 +85,14 @@ class CassandraExecutionDAOSpec extends CassandraSpec {
when:
// read the workflow from the datastore
- Workflow found = executionDAO.getWorkflow(workflowId)
+ WorkflowModel found = executionDAO.getWorkflow(workflowId)
then:
workflow == found
and:
// update the workflow
- workflow.setStatus(Workflow.WorkflowStatus.COMPLETED)
+ workflow.setStatus(WorkflowModel.Status.COMPLETED)
executionDAO.updateWorkflow(workflow)
when:
@@ -120,18 +120,18 @@ class CassandraExecutionDAOSpec extends CassandraSpec {
given: 'we create a workflow'
String workflowId = IDGenerator.generate()
WorkflowDef workflowDef = new WorkflowDef(name: 'def1', version: 1)
- Workflow workflow = new Workflow(workflowDefinition: workflowDef, workflowId: workflowId, input: new HashMap(), status: Workflow.WorkflowStatus.RUNNING, createTime: System.currentTimeMillis())
+ WorkflowModel workflow = new WorkflowModel(workflowDefinition: workflowDef, workflowId: workflowId, input: new HashMap(), status: WorkflowModel.Status.RUNNING, createTime: System.currentTimeMillis())
executionDAO.createWorkflow(workflow)
and: 'create tasks for this workflow'
- Task task1 = new Task(workflowInstanceId: workflowId, taskType: 'task1', referenceTaskName: 'task1', status: Task.Status.SCHEDULED, taskId: IDGenerator.generate())
- Task task2 = new Task(workflowInstanceId: workflowId, taskType: 'task2', referenceTaskName: 'task2', status: Task.Status.SCHEDULED, taskId: IDGenerator.generate())
- Task task3 = new Task(workflowInstanceId: workflowId, taskType: 'task3', referenceTaskName: 'task3', status: Task.Status.SCHEDULED, taskId: IDGenerator.generate())
+ TaskModel task1 = new TaskModel(workflowInstanceId: workflowId, taskType: 'task1', referenceTaskName: 'task1', status: TaskModel.Status.SCHEDULED, taskId: IDGenerator.generate())
+ TaskModel task2 = new TaskModel(workflowInstanceId: workflowId, taskType: 'task2', referenceTaskName: 'task2', status: TaskModel.Status.SCHEDULED, taskId: IDGenerator.generate())
+ TaskModel task3 = new TaskModel(workflowInstanceId: workflowId, taskType: 'task3', referenceTaskName: 'task3', status: TaskModel.Status.SCHEDULED, taskId: IDGenerator.generate())
def taskList = [task1, task2, task3]
when: 'add the tasks to the datastore'
- List
- * 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
- *
- * 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.
+ * Copyright 2022 Netflix, Inc.
+ *
+ * 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
+ *
+ * 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 com.netflix.conductor.cassandra.dao
import com.netflix.conductor.common.metadata.tasks.TaskDef
import com.netflix.conductor.common.metadata.workflow.WorkflowDef
+
import spock.lang.Subject
class CassandraMetadataDAOSpec extends CassandraSpec {
diff --git a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraSpec.groovy b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraSpec.groovy
index d9531f2b36..a5393210bb 100644
--- a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraSpec.groovy
+++ b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/dao/CassandraSpec.groovy
@@ -1,34 +1,35 @@
/*
- * Copyright 2021 Netflix, Inc.
- *
- * 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
- *
- * 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.
+ * Copyright 2022 Netflix, Inc.
+ *
+ * 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
+ *
+ * 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 com.netflix.conductor.cassandra.dao
-import com.datastax.driver.core.ConsistencyLevel
-import com.datastax.driver.core.Session
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.netflix.conductor.cassandra.config.CassandraProperties
-import com.netflix.conductor.cassandra.util.Statements
-import com.netflix.conductor.common.config.TestObjectMapperConfiguration
-import groovy.transform.PackageScope
+import java.time.Duration
+
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.context.ContextConfiguration
import org.testcontainers.containers.CassandraContainer
import org.testcontainers.spock.Testcontainers
+
+import com.netflix.conductor.cassandra.config.CassandraProperties
+import com.netflix.conductor.cassandra.util.Statements
+import com.netflix.conductor.common.config.TestObjectMapperConfiguration
+
+import com.datastax.driver.core.ConsistencyLevel
+import com.datastax.driver.core.Session
+import com.fasterxml.jackson.databind.ObjectMapper
+import groovy.transform.PackageScope
import spock.lang.Shared
import spock.lang.Specification
-import java.time.Duration
-
@ContextConfiguration(classes = [TestObjectMapperConfiguration.class])
@Testcontainers
@PackageScope
diff --git a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/util/StatementsSpec.groovy b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/util/StatementsSpec.groovy
index f674688b9e..f826a36208 100644
--- a/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/util/StatementsSpec.groovy
+++ b/cassandra-persistence/src/test/groovy/com/netflix/conductor/cassandra/util/StatementsSpec.groovy
@@ -1,16 +1,15 @@
/*
- * Copyright 2021 Netflix, Inc.
- *
- * 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
- *
- * 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.
+ * Copyright 2022 Netflix, Inc.
+ *
+ * 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
+ *
+ * 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 com.netflix.conductor.cassandra.util
import spock.lang.Specification
diff --git a/client/src/main/java/com/netflix/conductor/client/http/TaskClient.java b/client/src/main/java/com/netflix/conductor/client/http/TaskClient.java
index 3e7ca3699b..88700d2b83 100644
--- a/client/src/main/java/com/netflix/conductor/client/http/TaskClient.java
+++ b/client/src/main/java/com/netflix/conductor/client/http/TaskClient.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -248,12 +248,12 @@ public void evaluateAndUploadLargePayload(TaskResult taskResult, String taskType
MetricsContainer.recordTaskResultPayloadSize(taskType, taskResultSize);
long payloadSizeThreshold =
- conductorClientConfiguration.getTaskOutputPayloadThresholdKB() * 1024;
+ conductorClientConfiguration.getTaskOutputPayloadThresholdKB() * 1024L;
if (taskResultSize > payloadSizeThreshold) {
if (!conductorClientConfiguration.isExternalPayloadStorageEnabled()
|| taskResultSize
> conductorClientConfiguration.getTaskOutputMaxPayloadThresholdKB()
- * 1024) {
+ * 1024L) {
taskResult.setReasonForIncompletion(
String.format(
"The TaskResult payload size: %d is greater than the permissible %d MB",
diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java
index d2957f62b6..82a6a1af21 100644
--- a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java
+++ b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskResult.java b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskResult.java
index cf4e5c246c..d449176203 100644
--- a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskResult.java
+++ b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskResult.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/common/src/main/java/com/netflix/conductor/common/run/Workflow.java b/common/src/main/java/com/netflix/conductor/common/run/Workflow.java
index dc9d410820..946ab90ed4 100644
--- a/common/src/main/java/com/netflix/conductor/common/run/Workflow.java
+++ b/common/src/main/java/com/netflix/conductor/common/run/Workflow.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/dao/index/NoopIndexDAO.java b/contribs/src/main/java/com/netflix/conductor/contribs/dao/index/NoopIndexDAO.java
index dbbcfbd670..4b5f22c6b2 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/dao/index/NoopIndexDAO.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/dao/index/NoopIndexDAO.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -17,10 +17,10 @@
import java.util.concurrent.CompletableFuture;
import com.netflix.conductor.common.metadata.events.EventExecution;
-import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskExecLog;
import com.netflix.conductor.common.run.SearchResult;
-import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.common.run.TaskSummary;
+import com.netflix.conductor.common.run.WorkflowSummary;
import com.netflix.conductor.core.events.queue.Message;
import com.netflix.conductor.dao.IndexDAO;
@@ -34,18 +34,18 @@ public class NoopIndexDAO implements IndexDAO {
public void setup() {}
@Override
- public void indexWorkflow(Workflow workflow) {}
+ public void indexWorkflow(WorkflowSummary workflowSummary) {}
@Override
- public CompletableFuture
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWithTTLWorkflowStatusListener.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWithTTLWorkflowStatusListener.java
index e80bfcd2da..69b50d6a38 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWithTTLWorkflowStatusListener.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWithTTLWorkflowStatusListener.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -20,10 +20,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.WorkflowModel;
public class ArchivingWithTTLWorkflowStatusListener implements WorkflowStatusListener {
@@ -75,7 +75,7 @@ public void shutdownExecutorService() {
}
@Override
- public void onWorkflowCompleted(Workflow workflow) {
+ public void onWorkflowCompleted(WorkflowModel workflow) {
LOGGER.info("Archiving workflow {} on completion ", workflow.getWorkflowId());
if (delayArchiveSeconds > 0) {
scheduledThreadPoolExecutor.schedule(
@@ -90,7 +90,7 @@ public void onWorkflowCompleted(Workflow workflow) {
}
@Override
- public void onWorkflowTerminated(Workflow workflow) {
+ public void onWorkflowTerminated(WorkflowModel workflow) {
LOGGER.info("Archiving workflow {} on termination", workflow.getWorkflowId());
if (delayArchiveSeconds > 0) {
scheduledThreadPoolExecutor.schedule(
@@ -108,10 +108,10 @@ private class DelayArchiveWorkflow implements Runnable {
private final String workflowId;
private final String workflowName;
- private final Workflow.WorkflowStatus status;
+ private final WorkflowModel.Status status;
private final ExecutionDAOFacade executionDAOFacade;
- DelayArchiveWorkflow(Workflow workflow, ExecutionDAOFacade executionDAOFacade) {
+ DelayArchiveWorkflow(WorkflowModel workflow, ExecutionDAOFacade executionDAOFacade) {
this.workflowId = workflow.getWorkflowId();
this.workflowName = workflow.getWorkflowName();
this.status = workflow.getStatus();
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerConfiguration.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerConfiguration.java
index 482e63d739..c98c0c173b 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerConfiguration.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerConfiguration.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -17,8 +17,8 @@
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
@Configuration
@EnableConfigurationProperties(ArchivingWorkflowListenerProperties.class)
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerProperties.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerProperties.java
index 90076089ff..dfd57d3017 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerProperties.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowListenerProperties.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowStatusListener.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowStatusListener.java
index f84c99f705..f1fe98cca8 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowStatusListener.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/archive/ArchivingWorkflowStatusListener.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -15,10 +15,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.WorkflowModel;
/**
* Provides default implementation of workflow archiving immediately after workflow is completed or
@@ -37,14 +37,14 @@ public ArchivingWorkflowStatusListener(ExecutionDAOFacade executionDAOFacade) {
}
@Override
- public void onWorkflowCompleted(Workflow workflow) {
+ public void onWorkflowCompleted(WorkflowModel workflow) {
LOGGER.info("Archiving workflow {} on completion ", workflow.getWorkflowId());
this.executionDAOFacade.removeWorkflow(workflow.getWorkflowId(), true);
Monitors.recordWorkflowArchived(workflow.getWorkflowName(), workflow.getStatus());
}
@Override
- public void onWorkflowTerminated(Workflow workflow) {
+ public void onWorkflowTerminated(WorkflowModel workflow) {
LOGGER.info("Archiving workflow {} on termination", workflow.getWorkflowId());
this.executionDAOFacade.removeWorkflow(workflow.getWorkflowId(), true);
Monitors.recordWorkflowArchived(workflow.getWorkflowName(), workflow.getStatus());
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisher.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisher.java
index 45060f322d..c11ef06103 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisher.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisher.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -17,11 +17,12 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.common.run.WorkflowSummary;
+import com.netflix.conductor.core.dal.ModelMapper;
import com.netflix.conductor.core.events.queue.Message;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
import com.netflix.conductor.dao.QueueDAO;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -35,6 +36,7 @@ public class ConductorQueueStatusPublisher implements WorkflowStatusListener {
private static final Logger LOGGER =
LoggerFactory.getLogger(ConductorQueueStatusPublisher.class);
private final QueueDAO queueDAO;
+ private final ModelMapper modelMapper;
private final ObjectMapper objectMapper;
private final String successStatusQueue;
@@ -43,9 +45,11 @@ public class ConductorQueueStatusPublisher implements WorkflowStatusListener {
public ConductorQueueStatusPublisher(
QueueDAO queueDAO,
+ ModelMapper modelMapper,
ObjectMapper objectMapper,
ConductorQueueStatusPublisherProperties properties) {
this.queueDAO = queueDAO;
+ this.modelMapper = modelMapper;
this.objectMapper = objectMapper;
this.successStatusQueue = properties.getSuccessQueue();
this.failureStatusQueue = properties.getFailureQueue();
@@ -53,26 +57,26 @@ public ConductorQueueStatusPublisher(
}
@Override
- public void onWorkflowCompleted(Workflow workflow) {
+ public void onWorkflowCompleted(WorkflowModel workflow) {
LOGGER.info("Publishing callback of workflow {} on completion ", workflow.getWorkflowId());
queueDAO.push(successStatusQueue, Collections.singletonList(workflowToMessage(workflow)));
}
@Override
- public void onWorkflowTerminated(Workflow workflow) {
+ public void onWorkflowTerminated(WorkflowModel workflow) {
LOGGER.info("Publishing callback of workflow {} on termination", workflow.getWorkflowId());
queueDAO.push(failureStatusQueue, Collections.singletonList(workflowToMessage(workflow)));
}
@Override
- public void onWorkflowFinalized(Workflow workflow) {
+ public void onWorkflowFinalized(WorkflowModel workflow) {
LOGGER.info("Publishing callback of workflow {} on finalization", workflow.getWorkflowId());
queueDAO.push(finalizeStatusQueue, Collections.singletonList(workflowToMessage(workflow)));
}
- private Message workflowToMessage(Workflow workflow) {
+ private Message workflowToMessage(WorkflowModel workflow) {
String jsonWfSummary;
- WorkflowSummary summary = new WorkflowSummary(workflow);
+ WorkflowSummary summary = new WorkflowSummary(modelMapper.getWorkflow(workflow));
try {
jsonWfSummary = objectMapper.writeValueAsString(summary);
} catch (JsonProcessingException e) {
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherConfiguration.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherConfiguration.java
index 13d8d0a934..12dda6c444 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherConfiguration.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherConfiguration.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -17,6 +17,7 @@
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import com.netflix.conductor.core.dal.ModelMapper;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
import com.netflix.conductor.dao.QueueDAO;
@@ -32,8 +33,9 @@ public class ConductorQueueStatusPublisherConfiguration {
@Bean
public WorkflowStatusListener getWorkflowStatusListener(
QueueDAO queueDAO,
+ ModelMapper modelMapper,
ConductorQueueStatusPublisherProperties properties,
ObjectMapper objectMapper) {
- return new ConductorQueueStatusPublisher(queueDAO, objectMapper, properties);
+ return new ConductorQueueStatusPublisher(queueDAO, modelMapper, objectMapper, properties);
}
}
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherProperties.java b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherProperties.java
index e04d2fc2f0..ea9a53f743 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherProperties.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/listener/conductorqueue/ConductorQueueStatusPublisherProperties.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/SQSObservableQueue.java b/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/SQSObservableQueue.java
index 8ff19a0617..e4603791de 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/SQSObservableQueue.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/SQSObservableQueue.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/config/SQSEventQueueConfiguration.java b/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/config/SQSEventQueueConfiguration.java
index 6de5573159..412bbb8cb9 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/config/SQSEventQueueConfiguration.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/queue/sqs/config/SQSEventQueueConfiguration.java
@@ -32,7 +32,6 @@
import com.amazonaws.services.sqs.AmazonSQSClient;
import rx.Scheduler;
-@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Configuration
@EnableConfigurationProperties(SQSEventQueueProperties.class)
@ConditionalOnProperty(name = "conductor.event-queues.sqs.enabled", havingValue = "true")
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/storage/S3PayloadStorage.java b/contribs/src/main/java/com/netflix/conductor/contribs/storage/S3PayloadStorage.java
index eb9d7dd97d..e79bb6ebca 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/storage/S3PayloadStorage.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/storage/S3PayloadStorage.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -38,7 +38,9 @@
/**
* An implementation of {@link ExternalPayloadStorage} using AWS S3 for storing large JSON payload
- * data. The S3 client assumes that access to S3 is configured on the instance.
+ * data.
+ *
+ * NOTE: The S3 client assumes that access to S3 is configured on the instance.
*
* @see DefaultAWSCredentialsProviderChain
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/storage/config/S3Configuration.java b/contribs/src/main/java/com/netflix/conductor/contribs/storage/config/S3Configuration.java
index bb9ca31132..a52ec750cb 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/storage/config/S3Configuration.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/storage/config/S3Configuration.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/DefaultRestTemplateProvider.java b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/DefaultRestTemplateProvider.java
index 35a9a15e39..670f260638 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/DefaultRestTemplateProvider.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/DefaultRestTemplateProvider.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/HttpTask.java b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/HttpTask.java
index 2fce5ac87f..f0f42eaf4d 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/HttpTask.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/HttpTask.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -21,22 +21,17 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
+import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
-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.execution.WorkflowExecutor;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
import com.netflix.conductor.core.utils.Utils;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
@@ -79,12 +74,12 @@ public HttpTask(
}
@Override
- public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
+ public void start(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
Object request = task.getInputData().get(requestParameter);
task.setWorkerId(Utils.getServerId());
if (request == null) {
task.setReasonForIncompletion(MISSING_REQUEST);
- task.setStatus(Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
return;
}
@@ -93,14 +88,14 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
String reason =
"Missing HTTP URI. See documentation for HttpTask for required input parameters";
task.setReasonForIncompletion(reason);
- task.setStatus(Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
return;
}
if (input.getMethod() == null) {
String reason = "No HTTP method specified";
task.setReasonForIncompletion(reason);
- task.setStatus(Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
return;
}
@@ -113,9 +108,9 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
task.getTaskId());
if (response.statusCode > 199 && response.statusCode < 300) {
if (isAsyncComplete(task)) {
- task.setStatus(Status.IN_PROGRESS);
+ task.setStatus(TaskModel.Status.IN_PROGRESS);
} else {
- task.setStatus(Status.COMPLETED);
+ task.setStatus(TaskModel.Status.COMPLETED);
}
} else {
if (response.body != null) {
@@ -123,7 +118,7 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
} else {
task.setReasonForIncompletion("No response from the remote service");
}
- task.setStatus(Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
}
//noinspection ConstantConditions
if (response != null) {
@@ -139,7 +134,7 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
input.getVipAddress(),
task.getWorkflowInstanceId(),
e);
- task.setStatus(Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
task.setReasonForIncompletion(
"Failed to invoke " + getTaskType() + " task due to: " + e);
task.getOutputData().put("response", e.toString());
@@ -206,13 +201,13 @@ private Object extractBody(String responseBody) {
}
@Override
- public boolean execute(Workflow workflow, Task task, WorkflowExecutor executor) {
+ public boolean execute(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
return false;
}
@Override
- public void cancel(Workflow workflow, Task task, WorkflowExecutor executor) {
- task.setStatus(Status.CANCELED);
+ public void cancel(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
+ task.setStatus(TaskModel.Status.CANCELED);
}
@Override
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/RestTemplateProvider.java b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/RestTemplateProvider.java
index 665be8caab..95c648b079 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/RestTemplateProvider.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/http/RestTemplateProvider.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/json/JsonJqTransform.java b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/json/JsonJqTransform.java
index 9ec9f2f43d..26520edaf4 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/json/JsonJqTransform.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/json/JsonJqTransform.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -22,10 +22,10 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.execution.WorkflowExecutor;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -58,7 +58,7 @@ public JsonJqTransform(ObjectMapper objectMapper) {
}
@Override
- public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
+ public void start(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
final Map
* 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
diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/kafka/KafkaPublishTask.java b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/kafka/KafkaPublishTask.java
index 21f779ca6a..8ec91a0396 100644
--- a/contribs/src/main/java/com/netflix/conductor/contribs/tasks/kafka/KafkaPublishTask.java
+++ b/contribs/src/main/java/com/netflix/conductor/contribs/tasks/kafka/KafkaPublishTask.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -34,11 +34,11 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.execution.WorkflowExecutor;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
import com.netflix.conductor.core.utils.Utils;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
@@ -76,7 +76,7 @@ public KafkaPublishTask(KafkaProducerManager clientManager, ObjectMapper objectM
}
@Override
- public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
+ public void start(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
long taskStartMillis = Instant.now().toEpochMilli();
task.setWorkerId(Utils.getServerId());
@@ -109,9 +109,9 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
try {
recordMetaDataFuture.get();
if (isAsyncComplete(task)) {
- task.setStatus(Task.Status.IN_PROGRESS);
+ task.setStatus(TaskModel.Status.IN_PROGRESS);
} else {
- task.setStatus(Task.Status.COMPLETED);
+ task.setStatus(TaskModel.Status.COMPLETED);
}
long timeTakenToCompleteTask = Instant.now().toEpochMilli() - taskStartMillis;
LOGGER.debug("Published message {}, Time taken {}", input, timeTakenToCompleteTask);
@@ -133,9 +133,9 @@ public void start(Workflow workflow, Task task, WorkflowExecutor executor) {
}
}
- private void markTaskAsFailed(Task task, String reasonForIncompletion) {
+ private void markTaskAsFailed(TaskModel task, String reasonForIncompletion) {
task.setReasonForIncompletion(reasonForIncompletion);
- task.setStatus(Task.Status.FAILED);
+ task.setStatus(TaskModel.Status.FAILED);
}
/**
@@ -195,13 +195,13 @@ Object getKey(Input input) {
}
@Override
- public boolean execute(Workflow workflow, Task task, WorkflowExecutor executor) {
+ public boolean execute(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
return false;
}
@Override
- public void cancel(Workflow workflow, Task task, WorkflowExecutor executor) {
- task.setStatus(Task.Status.CANCELED);
+ public void cancel(WorkflowModel workflow, TaskModel task, WorkflowExecutor executor) {
+ task.setStatus(TaskModel.Status.CANCELED);
}
@Override
diff --git a/contribs/src/test/java/com/netflix/conductor/contribs/listener/ArchivingWorkflowStatusListenerTest.java b/contribs/src/test/java/com/netflix/conductor/contribs/listener/ArchivingWorkflowStatusListenerTest.java
index 3d58db5be5..9a89e86deb 100644
--- a/contribs/src/test/java/com/netflix/conductor/contribs/listener/ArchivingWorkflowStatusListenerTest.java
+++ b/contribs/src/test/java/com/netflix/conductor/contribs/listener/ArchivingWorkflowStatusListenerTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -19,24 +19,22 @@
import org.mockito.Mockito;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.contribs.listener.archive.ArchivingWorkflowStatusListener;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
+import com.netflix.conductor.model.WorkflowModel;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.*;
/** @author pavel.halabala */
public class ArchivingWorkflowStatusListenerTest {
- Workflow workflow;
+ WorkflowModel workflow;
ExecutionDAOFacade executionDAOFacade;
ArchivingWorkflowStatusListener listener;
@Before
public void before() {
- workflow = new Workflow();
+ workflow = new WorkflowModel();
WorkflowDef def = new WorkflowDef();
def.setName("name1");
def.setVersion(1);
diff --git a/contribs/src/test/java/com/netflix/conductor/contribs/queue/sqs/DefaultEventQueueProcessorTest.java b/contribs/src/test/java/com/netflix/conductor/contribs/queue/sqs/DefaultEventQueueProcessorTest.java
index 1b6c619db7..f661e62b33 100644
--- a/contribs/src/test/java/com/netflix/conductor/contribs/queue/sqs/DefaultEventQueueProcessorTest.java
+++ b/contribs/src/test/java/com/netflix/conductor/contribs/queue/sqs/DefaultEventQueueProcessorTest.java
@@ -30,6 +30,7 @@
import com.netflix.conductor.common.config.TestObjectMapperConfiguration;
import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.Task.Status;
+import com.netflix.conductor.common.metadata.tasks.TaskResult;
import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.events.queue.DefaultEventQueueProcessor;
import com.netflix.conductor.core.events.queue.Message;
@@ -63,7 +64,7 @@ public class DefaultEventQueueProcessorTest {
@Autowired private ObjectMapper objectMapper;
private static final List
* 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
diff --git a/contribs/src/test/java/com/netflix/conductor/contribs/tasks/http/HttpTaskTest.java b/contribs/src/test/java/com/netflix/conductor/contribs/tasks/http/HttpTaskTest.java
index f87eb6022b..065b492353 100644
--- a/contribs/src/test/java/com/netflix/conductor/contribs/tasks/http/HttpTaskTest.java
+++ b/contribs/src/test/java/com/netflix/conductor/contribs/tasks/http/HttpTaskTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -12,343 +12,356 @@
*/
package com.netflix.conductor.contribs.tasks.http;
-import org.junit.Ignore;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.*;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.model.MediaType;
+import org.testcontainers.containers.MockServerContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import com.netflix.conductor.common.metadata.tasks.TaskType;
+import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
+import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
+import com.netflix.conductor.core.execution.DeciderService;
+import com.netflix.conductor.core.execution.WorkflowExecutor;
+import com.netflix.conductor.core.execution.tasks.SystemTaskRegistry;
+import com.netflix.conductor.core.utils.ExternalPayloadStorageUtils;
+import com.netflix.conductor.core.utils.ParametersUtils;
+import com.netflix.conductor.dao.MetadataDAO;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
@SuppressWarnings("unchecked")
-@Ignore // Test causes "OutOfMemoryError" error during build
+// @Ignore // Test causes "OutOfMemoryError" error during build
public class HttpTaskTest {
- // private static final String ERROR_RESPONSE = "Something went wrong!";
- // private static final String TEXT_RESPONSE = "Text Response";
- // private static final double NUM_RESPONSE = 42.42d;
- //
- // private HttpTask httpTask;
- // private WorkflowExecutor workflowExecutor;
- // private final Workflow workflow = new Workflow();
- //
- // private static final ObjectMapper objectMapper = new ObjectMapper();
- // private static String JSON_RESPONSE;
- //
- // @ClassRule
- // public static MockServerContainer mockServer = new MockServerContainer(
- // DockerImageName.parse("mockserver/mockserver"));
- //
- // @BeforeClass
- // public static void init() throws Exception {
- // Map
* 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
@@ -19,8 +19,8 @@
import org.junit.Test;
import com.netflix.conductor.common.config.ObjectMapperProvider;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -35,8 +35,8 @@ public class JsonJqTransformTest {
@Test
public void dataShouldBeCorrectlySelected() {
final JsonJqTransform jsonJqTransform = new JsonJqTransform(objectMapper);
- final Workflow workflow = new Workflow();
- final Task task = new Task();
+ final WorkflowModel workflow = new WorkflowModel();
+ final TaskModel task = new TaskModel();
final Map
* 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
@@ -28,9 +28,9 @@
import org.springframework.test.context.junit4.SpringRunner;
import com.netflix.conductor.common.config.TestObjectMapperConfiguration;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.execution.WorkflowExecutor;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -50,15 +50,15 @@ public class KafkaPublishTaskTest {
public void missingRequest_Fail() {
KafkaPublishTask kafkaPublishTask =
new KafkaPublishTask(getKafkaProducerManager(), objectMapper);
- Task task = new Task();
- kafkaPublishTask.start(mock(Workflow.class), task, mock(WorkflowExecutor.class));
- assertEquals(Task.Status.FAILED, task.getStatus());
+ TaskModel task = new TaskModel();
+ kafkaPublishTask.start(mock(WorkflowModel.class), task, mock(WorkflowExecutor.class));
+ assertEquals(TaskModel.Status.FAILED, task.getStatus());
}
@Test
public void missingValue_Fail() {
- Task task = new Task();
+ TaskModel task = new TaskModel();
KafkaPublishTask.Input input = new KafkaPublishTask.Input();
input.setBootStrapServers("localhost:9092");
input.setTopic("testTopic");
@@ -67,14 +67,14 @@ public void missingValue_Fail() {
KafkaPublishTask kPublishTask =
new KafkaPublishTask(getKafkaProducerManager(), objectMapper);
- kPublishTask.start(mock(Workflow.class), task, mock(WorkflowExecutor.class));
- assertEquals(Task.Status.FAILED, task.getStatus());
+ kPublishTask.start(mock(WorkflowModel.class), task, mock(WorkflowExecutor.class));
+ assertEquals(TaskModel.Status.FAILED, task.getStatus());
}
@Test
public void missingBootStrapServers_Fail() {
- Task task = new Task();
+ TaskModel task = new TaskModel();
KafkaPublishTask.Input input = new KafkaPublishTask.Input();
Map
* 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
diff --git a/core/src/main/java/com/netflix/conductor/annotations/Trace.java b/core/src/main/java/com/netflix/conductor/annotations/Trace.java
index 8f59d23908..61da42cc47 100644
--- a/core/src/main/java/com/netflix/conductor/annotations/Trace.java
+++ b/core/src/main/java/com/netflix/conductor/annotations/Trace.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/LifecycleAwareComponent.java b/core/src/main/java/com/netflix/conductor/core/LifecycleAwareComponent.java
index 071eace191..bfe1455fb6 100644
--- a/core/src/main/java/com/netflix/conductor/core/LifecycleAwareComponent.java
+++ b/core/src/main/java/com/netflix/conductor/core/LifecycleAwareComponent.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/WorkflowContext.java b/core/src/main/java/com/netflix/conductor/core/WorkflowContext.java
index ab0208db54..2f996a47a6 100644
--- a/core/src/main/java/com/netflix/conductor/core/WorkflowContext.java
+++ b/core/src/main/java/com/netflix/conductor/core/WorkflowContext.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -12,7 +12,7 @@
*/
package com.netflix.conductor.core;
-/** Store the authentication context, app or user name or both */
+/** Store the authentication context, app or username or both */
public class WorkflowContext {
public static final ThreadLocal
* 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
@@ -10,7 +10,7 @@
* 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.orchestration;
+package com.netflix.conductor.core.dal;
import java.io.IOException;
import java.util.Collections;
@@ -32,23 +32,22 @@
import com.netflix.conductor.common.metadata.tasks.TaskDef;
import com.netflix.conductor.common.metadata.tasks.TaskExecLog;
import com.netflix.conductor.common.run.SearchResult;
+import com.netflix.conductor.common.run.TaskSummary;
import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.common.run.WorkflowSummary;
import com.netflix.conductor.core.config.ConductorProperties;
import com.netflix.conductor.core.events.queue.Message;
import com.netflix.conductor.core.exception.ApplicationException;
import com.netflix.conductor.core.exception.ApplicationException.Code;
-import com.netflix.conductor.dao.ConcurrentExecutionLimitDAO;
-import com.netflix.conductor.dao.ExecutionDAO;
-import com.netflix.conductor.dao.IndexDAO;
-import com.netflix.conductor.dao.PollDataDAO;
-import com.netflix.conductor.dao.QueueDAO;
-import com.netflix.conductor.dao.RateLimitingDAO;
+import com.netflix.conductor.dao.*;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
-import static com.netflix.conductor.core.execution.WorkflowExecutor.DECIDER_QUEUE;
+import static com.netflix.conductor.core.utils.Utils.DECIDER_QUEUE;
/**
* Service that acts as a facade for accessing execution data from the {@link ExecutionDAO}, {@link
@@ -69,6 +68,7 @@ public class ExecutionDAOFacade {
private final RateLimitingDAO rateLimitingDao;
private final ConcurrentExecutionLimitDAO concurrentExecutionLimitDAO;
private final PollDataDAO pollDataDAO;
+ private final ModelMapper modelMapper;
private final ObjectMapper objectMapper;
private final ConductorProperties properties;
@@ -81,6 +81,7 @@ public ExecutionDAOFacade(
RateLimitingDAO rateLimitingDao,
ConcurrentExecutionLimitDAO concurrentExecutionLimitDAO,
PollDataDAO pollDataDAO,
+ ModelMapper modelMapper,
ObjectMapper objectMapper,
ConductorProperties properties) {
this.executionDAO = executionDAO;
@@ -89,6 +90,7 @@ public ExecutionDAOFacade(
this.rateLimitingDao = rateLimitingDao;
this.concurrentExecutionLimitDAO = concurrentExecutionLimitDAO;
this.pollDataDAO = pollDataDAO;
+ this.modelMapper = modelMapper;
this.objectMapper = objectMapper;
this.properties = properties;
this.scheduledThreadPoolExecutor =
@@ -126,6 +128,10 @@ public void shutdownExecutorService() {
}
}
+ public WorkflowModel getWorkflowModel(String workflowId, boolean includeTasks) {
+ return modelMapper.getFullCopy(getWorkflowFromDatastore(workflowId, includeTasks));
+ }
+
/**
* Fetches the {@link Workflow} object from the data store given the id. Attempts to fetch from
* {@link ExecutionDAO} first, if not found, attempts to fetch from {@link IndexDAO}.
@@ -139,8 +145,12 @@ public void shutdownExecutorService() {
*
+ * 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
+ *
+ * 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 com.netflix.conductor.core.dal;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Component;
+
+import com.netflix.conductor.common.metadata.tasks.Task;
+import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.common.utils.ExternalPayloadStorage.Operation;
+import com.netflix.conductor.common.utils.ExternalPayloadStorage.PayloadType;
+import com.netflix.conductor.core.utils.ExternalPayloadStorageUtils;
+import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
+
+@Component
+public class ModelMapper {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ModelMapper.class);
+
+ private final ExternalPayloadStorageUtils externalPayloadStorageUtils;
+
+ public ModelMapper(ExternalPayloadStorageUtils externalPayloadStorageUtils) {
+ this.externalPayloadStorageUtils = externalPayloadStorageUtils;
+ }
+
+ /**
+ * Fetch the fully formed workflow domain object with complete payloads
+ *
+ * @param workflowModel the workflow domain object from the datastore
+ * @return the workflow domain object {@link WorkflowModel} with payloads from external storage
+ */
+ public WorkflowModel getFullCopy(WorkflowModel workflowModel) {
+ populateWorkflowAndTaskPayloadData(workflowModel);
+ return workflowModel;
+ }
+
+ /**
+ * Get a lean copy of the workflow domain object with large payloads externalized
+ *
+ * @param workflowModel the fully formed workflow domain object
+ * @return the workflow domain object {@link WorkflowModel} with large payloads externalized
+ */
+ public WorkflowModel getLeanCopy(WorkflowModel workflowModel) {
+ WorkflowModel leanWorkflowModel = workflowModel.copy();
+ externalizeWorkflowData(leanWorkflowModel);
+ workflowModel.getTasks().forEach(this::getLeanCopy);
+ return leanWorkflowModel;
+ }
+
+ /**
+ * Map the workflow domain object to the workflow DTO
+ *
+ * @param workflowModel the workflow domain object {@link WorkflowModel}
+ * @return the workflow DTO {@link Workflow}
+ */
+ public Workflow getWorkflow(WorkflowModel workflowModel) {
+ externalizeWorkflowData(workflowModel);
+
+ Workflow workflow = new Workflow();
+ BeanUtils.copyProperties(workflowModel, workflow);
+ workflow.setStatus(Workflow.WorkflowStatus.valueOf(workflowModel.getStatus().name()));
+ workflow.setTasks(
+ workflowModel.getTasks().stream().map(this::getTask).collect(Collectors.toList()));
+
+ return workflow;
+ }
+
+ /**
+ * Fetch the fully formed task domain object with complete payloads
+ *
+ * @param taskModel the task domain object from the datastore
+ * @return the task domain object {@link TaskModel} with payloads from external storage
+ */
+ public TaskModel getFullCopy(TaskModel taskModel) {
+ populateTaskData(taskModel);
+ return taskModel;
+ }
+
+ /**
+ * Get a lean copy of the task domain object with large payloads externalized
+ *
+ * @param taskModel the fully formed task domain object
+ * @return the task domain object {@link TaskModel} with large payloads externalized
+ */
+ public TaskModel getLeanCopy(TaskModel taskModel) {
+ TaskModel leanTaskModel = taskModel.copy();
+ externalizeTaskData(leanTaskModel);
+ return leanTaskModel;
+ }
+
+ /**
+ * Map the task domain object to the task DTO
+ *
+ * @param taskModel the task domain object {@link TaskModel}
+ * @return the task DTO {@link Task}
+ */
+ public Task getTask(TaskModel taskModel) {
+ externalizeTaskData(taskModel);
+
+ Task task = new Task();
+ BeanUtils.copyProperties(taskModel, task);
+ task.setStatus(Task.Status.valueOf(taskModel.getStatus().name()));
+ return task;
+ }
+
+ /**
+ * Populates the workflow input data and the tasks input/output data if stored in external
+ * payload storage.
+ *
+ * @param workflowModel the workflowModel for which the payload data needs to be populated from
+ * external storage (if applicable)
+ */
+ private void populateWorkflowAndTaskPayloadData(WorkflowModel workflowModel) {
+ if (StringUtils.isNotBlank(workflowModel.getExternalInputPayloadStoragePath())) {
+ Map
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/DefaultEventProcessor.java b/core/src/main/java/com/netflix/conductor/core/events/DefaultEventProcessor.java
index 309fd8c660..52e50f288a 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/DefaultEventProcessor.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/DefaultEventProcessor.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/DefaultEventQueueManager.java b/core/src/main/java/com/netflix/conductor/core/events/DefaultEventQueueManager.java
index d71cbda82a..1c83096586 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/DefaultEventQueueManager.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/DefaultEventQueueManager.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -49,7 +49,6 @@
*
* @see DefaultEventQueueProcessor
*/
-@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Component
@ConditionalOnProperty(
name = "conductor.default-event-processor.enabled",
diff --git a/core/src/main/java/com/netflix/conductor/core/events/EventQueueManager.java b/core/src/main/java/com/netflix/conductor/core/events/EventQueueManager.java
index 9b532c38d8..fc6a568f68 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/EventQueueManager.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/EventQueueManager.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/EventQueueProvider.java b/core/src/main/java/com/netflix/conductor/core/events/EventQueueProvider.java
index 4bcceb0b8c..8bd11f929a 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/EventQueueProvider.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/EventQueueProvider.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/EventQueues.java b/core/src/main/java/com/netflix/conductor/core/events/EventQueues.java
index e29b17ee5b..b4ab2e388f 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/EventQueues.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/EventQueues.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/ScriptEvaluator.java b/core/src/main/java/com/netflix/conductor/core/events/ScriptEvaluator.java
index 44e3aaae6b..506ded2211 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/ScriptEvaluator.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/ScriptEvaluator.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/SimpleActionProcessor.java b/core/src/main/java/com/netflix/conductor/core/events/SimpleActionProcessor.java
index 0f421f6da0..7265e21dce 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/SimpleActionProcessor.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/SimpleActionProcessor.java
@@ -24,14 +24,14 @@
import com.netflix.conductor.common.metadata.events.EventHandler.Action;
import com.netflix.conductor.common.metadata.events.EventHandler.StartWorkflow;
import com.netflix.conductor.common.metadata.events.EventHandler.TaskDetails;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.metadata.tasks.Task.Status;
import com.netflix.conductor.common.metadata.tasks.TaskResult;
-import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.core.dal.ModelMapper;
import com.netflix.conductor.core.execution.WorkflowExecutor;
import com.netflix.conductor.core.utils.JsonUtils;
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
/**
* Action Processor subscribes to the Event Actions queue and processes the actions (e.g. start
@@ -43,14 +43,17 @@ public class SimpleActionProcessor implements ActionProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(SimpleActionProcessor.class);
private final WorkflowExecutor workflowExecutor;
+ private final ModelMapper modelMapper;
private final ParametersUtils parametersUtils;
private final JsonUtils jsonUtils;
public SimpleActionProcessor(
WorkflowExecutor workflowExecutor,
+ ModelMapper modelMapper,
ParametersUtils parametersUtils,
JsonUtils jsonUtils) {
this.workflowExecutor = workflowExecutor;
+ this.modelMapper = modelMapper;
this.parametersUtils = parametersUtils;
this.jsonUtils = jsonUtils;
}
@@ -77,12 +80,17 @@ public Map
* 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
@@ -35,7 +35,6 @@
*
* @see ConductorObservableQueue
*/
-@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Component
@ConditionalOnProperty(
name = "conductor.event-queues.default.enabled",
diff --git a/core/src/main/java/com/netflix/conductor/core/events/queue/ConductorObservableQueue.java b/core/src/main/java/com/netflix/conductor/core/events/queue/ConductorObservableQueue.java
index 4ecf0ea875..649cc7b501 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/queue/ConductorObservableQueue.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/queue/ConductorObservableQueue.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/queue/DefaultEventQueueProcessor.java b/core/src/main/java/com/netflix/conductor/core/events/queue/DefaultEventQueueProcessor.java
index 7255b51e99..de3a2811b8 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/queue/DefaultEventQueueProcessor.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/queue/DefaultEventQueueProcessor.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -27,6 +27,7 @@
import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.Task.Status;
+import com.netflix.conductor.common.metadata.tasks.TaskResult;
import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.exception.ApplicationException;
import com.netflix.conductor.core.exception.ApplicationException.Code;
@@ -55,8 +56,7 @@ public class DefaultEventQueueProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultEventQueueProcessor.class);
private final Map
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/events/queue/ObservableQueue.java b/core/src/main/java/com/netflix/conductor/core/events/queue/ObservableQueue.java
index c9195b6f63..897090ba7f 100644
--- a/core/src/main/java/com/netflix/conductor/core/events/queue/ObservableQueue.java
+++ b/core/src/main/java/com/netflix/conductor/core/events/queue/ObservableQueue.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
diff --git a/core/src/main/java/com/netflix/conductor/core/exception/ApplicationException.java b/core/src/main/java/com/netflix/conductor/core/exception/ApplicationException.java
index baa0b11657..f1a086d403 100644
--- a/core/src/main/java/com/netflix/conductor/core/exception/ApplicationException.java
+++ b/core/src/main/java/com/netflix/conductor/core/exception/ApplicationException.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2020 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -18,7 +18,6 @@
import java.util.LinkedHashMap;
import java.util.Map;
-@SuppressWarnings("serial")
public class ApplicationException extends RuntimeException {
public enum Code {
@@ -78,7 +77,7 @@ public String getTrace() {
PrintStream ps = new PrintStream(baos);
this.printStackTrace(ps);
ps.flush();
- return new String(baos.toByteArray());
+ return baos.toString();
}
public Map
* 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
@@ -12,33 +12,36 @@
*/
package com.netflix.conductor.core.exception;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow.WorkflowStatus;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
+
+import static com.netflix.conductor.model.WorkflowModel.Status.FAILED;
public class TerminateWorkflowException extends RuntimeException {
- private final WorkflowStatus workflowStatus;
- private final Task task;
+ private final WorkflowModel.Status workflowStatus;
+ private final TaskModel task;
public TerminateWorkflowException(String reason) {
- this(reason, WorkflowStatus.FAILED);
+ this(reason, FAILED);
}
- public TerminateWorkflowException(String reason, WorkflowStatus workflowStatus) {
+ public TerminateWorkflowException(String reason, WorkflowModel.Status workflowStatus) {
this(reason, workflowStatus, null);
}
- public TerminateWorkflowException(String reason, WorkflowStatus workflowStatus, Task task) {
+ public TerminateWorkflowException(
+ String reason, WorkflowModel.Status workflowStatus, TaskModel task) {
super(reason);
this.workflowStatus = workflowStatus;
this.task = task;
}
- public WorkflowStatus getWorkflowStatus() {
+ public WorkflowModel.Status getWorkflowStatus() {
return workflowStatus;
}
- public Task getTask() {
+ public TaskModel getTask() {
return task;
}
}
diff --git a/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java b/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
index 07afb03672..cb5d737d35 100644
--- a/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -16,19 +16,15 @@
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.run.Workflow;
import com.netflix.conductor.core.config.ConductorProperties;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
import com.netflix.conductor.core.utils.QueueUtils;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.metrics.Monitors;
-
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.CANCELED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.IN_PROGRESS;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.SCHEDULED;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
@Component
public class AsyncSystemTaskExecutor {
@@ -39,7 +35,6 @@ public class AsyncSystemTaskExecutor {
private final long queueTaskMessagePostponeSecs;
private final long systemTaskCallbackTime;
private final WorkflowExecutor workflowExecutor;
- private final DeciderService deciderService;
private static final Logger LOGGER = LoggerFactory.getLogger(AsyncSystemTaskExecutor.class);
@@ -48,13 +43,11 @@ public AsyncSystemTaskExecutor(
QueueDAO queueDAO,
MetadataDAO metadataDAO,
ConductorProperties conductorProperties,
- WorkflowExecutor workflowExecutor,
- DeciderService deciderService) {
+ WorkflowExecutor workflowExecutor) {
this.executionDAOFacade = executionDAOFacade;
this.queueDAO = queueDAO;
this.metadataDAO = metadataDAO;
this.workflowExecutor = workflowExecutor;
- this.deciderService = deciderService;
this.systemTaskCallbackTime =
conductorProperties.getSystemTaskWorkerCallbackDuration().getSeconds();
this.queueTaskMessagePostponeSecs =
@@ -65,10 +58,10 @@ public AsyncSystemTaskExecutor(
* Executes and persists the results of an async {@link WorkflowSystemTask}.
*
* @param systemTask The {@link WorkflowSystemTask} to be executed.
- * @param taskId The id of the {@link Task} object.
+ * @param taskId The id of the {@link TaskModel} object.
*/
public void execute(WorkflowSystemTask systemTask, String taskId) {
- Task task = loadTaskQuietly(taskId);
+ TaskModel task = loadTaskQuietly(taskId);
if (task == null) {
LOGGER.error("TaskId: {} could not be found while executing {}", taskId, systemTask);
return;
@@ -84,9 +77,8 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
return;
}
- if (task.getStatus().equals(SCHEDULED)) {
+ if (task.getStatus().equals(TaskModel.Status.SCHEDULED)) {
if (executionDAOFacade.exceedsInProgressLimit(task)) {
- // TODO: add a metric to record this
LOGGER.warn(
"Concurrent Execution limited for {}:{}", taskId, task.getTaskDefName());
postponeQuietly(queueName, task);
@@ -110,7 +102,7 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
// if we are here the Task object is updated and needs to be persisted regardless of an
// exception
try {
- Workflow workflow = executionDAOFacade.getWorkflowById(workflowId, true);
+ WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
if (workflow.getStatus().isTerminal()) {
LOGGER.info(
@@ -119,7 +111,7 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
systemTask,
task.getTaskId());
if (!task.getStatus().isTerminal()) {
- task.setStatus(CANCELED);
+ task.setStatus(TaskModel.Status.CANCELED);
task.setReasonForIncompletion(
String.format(
"Workflow is in %s state", workflow.getStatus().toString()));
@@ -134,29 +126,22 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
task.getTaskId(),
task.getStatus());
- // load task data (input/output) from external storage, if necessary
- deciderService.populateTaskData(task);
-
boolean isTaskAsyncComplete = systemTask.isAsyncComplete(task);
- if (task.getStatus() == SCHEDULED || !isTaskAsyncComplete) {
+ if (task.getStatus() == TaskModel.Status.SCHEDULED || !isTaskAsyncComplete) {
task.incrementPollCount();
}
- if (task.getStatus() == SCHEDULED) {
+ if (task.getStatus() == TaskModel.Status.SCHEDULED) {
task.setStartTime(System.currentTimeMillis());
Monitors.recordQueueWaitTime(task.getTaskDefName(), task.getQueueWaitTime());
systemTask.start(workflow, task, workflowExecutor);
- } else if (task.getStatus() == IN_PROGRESS) {
+ } else if (task.getStatus() == TaskModel.Status.IN_PROGRESS) {
systemTask.execute(workflow, task, workflowExecutor);
}
- if (task.getOutputData() != null && !task.getOutputData().isEmpty()) {
- deciderService.externalizeTaskData(task);
- }
-
// Update message in Task queue based on Task status
// Remove asyncComplete system tasks from the queue that are not in SCHEDULED state
- if (isTaskAsyncComplete && task.getStatus() != SCHEDULED) {
+ if (isTaskAsyncComplete && task.getStatus() != TaskModel.Status.SCHEDULED) {
queueDAO.remove(queueName, task.getTaskId());
hasTaskExecutionCompleted = true;
} else if (task.getStatus().isTerminal()) {
@@ -191,7 +176,7 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
}
}
- private void postponeQuietly(String queueName, Task task) {
+ private void postponeQuietly(String queueName, TaskModel task) {
try {
queueDAO.postpone(
queueName,
@@ -203,9 +188,9 @@ private void postponeQuietly(String queueName, Task task) {
}
}
- private Task loadTaskQuietly(String taskId) {
+ private TaskModel loadTaskQuietly(String taskId) {
try {
- return executionDAOFacade.getTaskById(taskId);
+ return executionDAOFacade.getTaskModel(taskId);
} catch (Exception e) {
return null;
}
diff --git a/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java b/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java
index 9f04f1f51e..66cc2c9e74 100644
--- a/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java
+++ b/core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 Netflix, Inc.
+ * Copyright 2022 Netflix, Inc.
*
* 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
@@ -13,14 +13,7 @@
package com.netflix.conductor.core.execution;
import java.time.Duration;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -31,14 +24,10 @@
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.metadata.tasks.Task.Status;
import com.netflix.conductor.common.metadata.tasks.TaskDef;
import com.netflix.conductor.common.metadata.tasks.TaskType;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
-import com.netflix.conductor.common.run.Workflow;
-import com.netflix.conductor.common.run.Workflow.WorkflowStatus;
import com.netflix.conductor.common.utils.ExternalPayloadStorage.Operation;
import com.netflix.conductor.common.utils.ExternalPayloadStorage.PayloadType;
import com.netflix.conductor.common.utils.TaskUtils;
@@ -51,22 +40,19 @@
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.google.common.annotations.VisibleForTesting;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.COMPLETED_WITH_ERRORS;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.IN_PROGRESS;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.SCHEDULED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.SKIPPED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.TIMED_OUT;
import static com.netflix.conductor.common.metadata.tasks.TaskType.TERMINATE;
+import static com.netflix.conductor.model.TaskModel.Status.*;
/**
* Decider evaluates the state of the workflow by inspecting the current state along with the
* blueprint. The result of the evaluation is either to schedule further tasks, complete/fail the
* workflow or do nothing.
*/
-@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@Service
public class DeciderService {
@@ -82,10 +68,10 @@ public class DeciderService {
private final Map
* 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
@@ -12,16 +12,7 @@
*/
package com.netflix.conductor.core.execution;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -31,23 +22,17 @@
import org.springframework.stereotype.Component;
import com.netflix.conductor.annotations.Trace;
-import com.netflix.conductor.common.metadata.tasks.PollData;
-import com.netflix.conductor.common.metadata.tasks.Task;
-import com.netflix.conductor.common.metadata.tasks.TaskDef;
-import com.netflix.conductor.common.metadata.tasks.TaskExecLog;
-import com.netflix.conductor.common.metadata.tasks.TaskResult;
-import com.netflix.conductor.common.metadata.tasks.TaskResult.Status;
-import com.netflix.conductor.common.metadata.tasks.TaskType;
+import com.netflix.conductor.common.metadata.tasks.*;
import com.netflix.conductor.common.metadata.workflow.RerunWorkflowRequest;
import com.netflix.conductor.common.metadata.workflow.SkipTaskRequest;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
import com.netflix.conductor.common.run.Workflow;
-import com.netflix.conductor.common.run.Workflow.WorkflowStatus;
import com.netflix.conductor.common.utils.RetryUtil;
import com.netflix.conductor.common.utils.TaskUtils;
import com.netflix.conductor.core.WorkflowContext;
import com.netflix.conductor.core.config.ConductorProperties;
+import com.netflix.conductor.core.dal.ExecutionDAOFacade;
import com.netflix.conductor.core.exception.ApplicationException;
import com.netflix.conductor.core.exception.ApplicationException.Code;
import com.netflix.conductor.core.exception.TerminateWorkflowException;
@@ -56,33 +41,22 @@
import com.netflix.conductor.core.execution.tasks.WorkflowSystemTask;
import com.netflix.conductor.core.listener.WorkflowStatusListener;
import com.netflix.conductor.core.metadata.MetadataMapperService;
-import com.netflix.conductor.core.orchestration.ExecutionDAOFacade;
import com.netflix.conductor.core.utils.IDGenerator;
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.core.utils.QueueUtils;
+import com.netflix.conductor.core.utils.Utils;
import com.netflix.conductor.dao.MetadataDAO;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.metrics.Monitors;
+import com.netflix.conductor.model.TaskModel;
+import com.netflix.conductor.model.WorkflowModel;
import com.netflix.conductor.service.ExecutionLockService;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.CANCELED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.FAILED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.FAILED_WITH_TERMINAL_ERROR;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.IN_PROGRESS;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.SCHEDULED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.SKIPPED;
-import static com.netflix.conductor.common.metadata.tasks.Task.Status.valueOf;
-import static com.netflix.conductor.common.metadata.tasks.TaskType.TASK_TYPE_FORK_JOIN_DYNAMIC;
-import static com.netflix.conductor.common.metadata.tasks.TaskType.TASK_TYPE_JOIN;
-import static com.netflix.conductor.common.metadata.tasks.TaskType.TASK_TYPE_SUB_WORKFLOW;
-import static com.netflix.conductor.common.metadata.tasks.TaskType.TERMINATE;
-import static com.netflix.conductor.core.exception.ApplicationException.Code.BACKEND_ERROR;
-import static com.netflix.conductor.core.exception.ApplicationException.Code.CONFLICT;
-import static com.netflix.conductor.core.exception.ApplicationException.Code.INVALID_INPUT;
-import static com.netflix.conductor.core.exception.ApplicationException.Code.NOT_FOUND;
+import static com.netflix.conductor.core.exception.ApplicationException.Code.*;
+import static com.netflix.conductor.model.TaskModel.Status.*;
/** Workflow services provider interface */
@Trace
@@ -103,23 +77,23 @@ public class WorkflowExecutor {
private final SystemTaskRegistry systemTaskRegistry;
private long activeWorkerLastPollMs;
- public static final String DECIDER_QUEUE = "_deciderQueue";
private static final String CLASS_NAME = WorkflowExecutor.class.getSimpleName();
private final ExecutionLockService executionLockService;
- private static final Predicate