diff --git a/.gitignore b/.gitignore index a1bedb5ac6..eb2ac67732 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ ui/dist ui/package-lock.json .gradle .project +bin build client/python/conductor.egg-info *.pyc @@ -19,3 +20,4 @@ out/ bin/ target/ .DS_Store +target/ diff --git a/client/src/test/java/com/netflix/conductor/client/http/MetadataClientTest.java b/client/src/test/java/com/netflix/conductor/client/http/MetadataClientTest.java index d4e9083413..4a5948b989 100644 --- a/client/src/test/java/com/netflix/conductor/client/http/MetadataClientTest.java +++ b/client/src/test/java/com/netflix/conductor/client/http/MetadataClientTest.java @@ -1,22 +1,16 @@ package com.netflix.conductor.client.http; -import com.sun.jersey.api.client.Client; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.mockito.Mockito; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; -import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import com.netflix.conductor.client.http.MetadataClient; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.ExpectedException; -import org.mockito.Mockito; -import org.mockito.MockitoAnnotations; /** diff --git a/client/src/test/java/com/netflix/conductor/client/metadata/workflow/TestWorkflowTask.java b/client/src/test/java/com/netflix/conductor/client/metadata/workflow/TestWorkflowTask.java index dbd4868036..e988aeabac 100644 --- a/client/src/test/java/com/netflix/conductor/client/metadata/workflow/TestWorkflowTask.java +++ b/client/src/test/java/com/netflix/conductor/client/metadata/workflow/TestWorkflowTask.java @@ -1,12 +1,12 @@ -/** +/* * Copyright 2016 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 - * + *

+ * 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. @@ -15,46 +15,65 @@ */ package com.netflix.conductor.client.metadata.workflow; -import static org.junit.Assert.*; - +import com.fasterxml.jackson.databind.ObjectMapper; +import com.netflix.conductor.common.metadata.tasks.Task; import com.netflix.conductor.common.metadata.workflow.TaskType; +import com.netflix.conductor.common.metadata.workflow.WorkflowTask; +import com.netflix.conductor.common.utils.JsonMapperProvider; +import org.junit.Before; import org.junit.Test; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.netflix.conductor.common.metadata.workflow.WorkflowTask; +import java.io.InputStream; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; /** - * * @author Viren - * */ public class TestWorkflowTask { - @Test - public void test() throws Exception { - ObjectMapper om = new ObjectMapper(); - WorkflowTask task = new WorkflowTask(); - task.setType("Hello"); - task.setName("name"); - - String json = om.writeValueAsString(task); - - WorkflowTask read = om.readValue(json, WorkflowTask.class); - assertNotNull(read); - assertEquals(task.getName(), read.getName()); - assertEquals(task.getType(), read.getType()); - - task = new WorkflowTask(); - task.setWorkflowTaskType(TaskType.SUB_WORKFLOW); - task.setName("name"); - - json = om.writeValueAsString(task); - - read = om.readValue(json, WorkflowTask.class); - assertNotNull(read); - assertEquals(task.getName(), read.getName()); - assertEquals(task.getType(), read.getType()); - assertEquals(TaskType.SUB_WORKFLOW.name(), read.getType()); - } + private ObjectMapper objectMapper; + + @Before + public void setup() { + objectMapper = new JsonMapperProvider().get(); + } + + @Test + public void test() throws Exception { + WorkflowTask task = new WorkflowTask(); + task.setType("Hello"); + task.setName("name"); + + String json = objectMapper.writeValueAsString(task); + + WorkflowTask read = objectMapper.readValue(json, WorkflowTask.class); + assertNotNull(read); + assertEquals(task.getName(), read.getName()); + assertEquals(task.getType(), read.getType()); + + task = new WorkflowTask(); + task.setWorkflowTaskType(TaskType.SUB_WORKFLOW); + task.setName("name"); + + json = objectMapper.writeValueAsString(task); + + read = objectMapper.readValue(json, WorkflowTask.class); + assertNotNull(read); + assertEquals(task.getName(), read.getName()); + assertEquals(task.getType(), read.getType()); + assertEquals(TaskType.SUB_WORKFLOW.name(), read.getType()); + } + @SuppressWarnings("unchecked") + @Test + public void testObectMapper() throws Exception { + try (InputStream stream = TestWorkflowTask.class.getResourceAsStream("/tasks.json")) { + List tasks = objectMapper.readValue(stream, List.class); + assertNotNull(tasks); + assertEquals(1, tasks.size()); + } + } } diff --git a/client/src/test/resources/tasks.json b/client/src/test/resources/tasks.json new file mode 100644 index 0000000000..424b4880ec --- /dev/null +++ b/client/src/test/resources/tasks.json @@ -0,0 +1,70 @@ +[ + { + "taskType": "task_1", + "status": "IN_PROGRESS", + "inputData": { + "mod": null, + "oddEven": null + }, + "referenceTaskName": "task_1", + "retryCount": 0, + "seq": 1, + "pollCount": 1, + "taskDefName": "task_1", + "scheduledTime": 1539623183131, + "startTime": 1539623436841, + "endTime": 0, + "updateTime": 1539623436841, + "startDelayInSeconds": 0, + "retried": false, + "executed": false, + "callbackFromWorker": true, + "responseTimeoutSeconds": 0, + "workflowInstanceId": "2d525ed8-d0e5-44c8-a2df-a110b25c09ac", + "workflowType": "kitchensink", + "taskId": "bc5d9deb-cf86-443d-a1f6-59c36d2464f7", + "callbackAfterSeconds": 0, + "workerId": "test", + "workflowTask": { + "name": "task_1", + "taskReferenceName": "task_1", + "inputParameters": { + "mod": "${workflow.input.mod}", + "oddEven": "${workflow.input.oddEven}" + }, + "type": "SIMPLE", + "startDelay": 0, + "optional": false, + "taskDefinition": { + "ownerApp": "falguni-test", + "createTime": 1534274994644, + "createdBy": "CPEWORKFLOW", + "name": "task_1", + "description": "Test Task 01", + "retryCount": 0, + "timeoutSeconds": 5, + "inputKeys": [ + "mod", + "oddEven" + ], + "outputKeys": [ + "someOutput" + ], + "timeoutPolicy": "TIME_OUT_WF", + "retryLogic": "FIXED", + "retryDelaySeconds": 0, + "responseTimeoutSeconds": 0, + "concurrentExecLimit": 0, + "rateLimitPerFrequency": 0, + "rateLimitFrequencyInSeconds": 1 + } + }, + "rateLimitPerFrequency": 0, + "rateLimitFrequencyInSeconds": 0, + "taskDefinition": { + "present": true + }, + "queueWaitTime": 253710, + "taskStatus": "IN_PROGRESS" + } +] \ No newline at end of file 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 0aae4b857e..1210a732a5 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,4 +1,4 @@ -/** +/* * Copyright 2016 Netflix, Inc. *

* Licensed under the Apache License, Version 2.0 (the "License"); @@ -135,7 +135,7 @@ public boolean isRetriable() { private boolean callbackFromWorker = true; @ProtoField(id = 19) - private int responseTimeoutSeconds; + private long responseTimeoutSeconds; @ProtoField(id = 20) private String workflowInstanceId; @@ -459,14 +459,14 @@ public void setTaskDefName(String taskDefName) { /** * @return the timeout for task to send response. After this timeout, the task will be re-queued */ - public int getResponseTimeoutSeconds() { + public long getResponseTimeoutSeconds() { return responseTimeoutSeconds; } /** * @param responseTimeoutSeconds - timeout for task to send response. After this timeout, the task will be re-queued */ - public void setResponseTimeoutSeconds(int responseTimeoutSeconds) { + public void setResponseTimeoutSeconds(long responseTimeoutSeconds) { this.responseTimeoutSeconds = responseTimeoutSeconds; } diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskDef.java b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskDef.java index 4a372a2898..7ede7b426b 100644 --- a/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskDef.java +++ b/common/src/main/java/com/netflix/conductor/common/metadata/tasks/TaskDef.java @@ -74,7 +74,7 @@ public static enum RetryLogic {FIXED, EXPONENTIAL_BACKOFF} private int retryDelaySeconds = 60; @ProtoField(id = 10) - private int responseTimeoutSeconds = ONE_HOUR; + private long responseTimeoutSeconds = ONE_HOUR; @ProtoField(id = 11) private Integer concurrentExecLimit; @@ -236,7 +236,7 @@ public int getRetryDelaySeconds() { * * @return the timeout for task to send response. After this timeout, the task will be re-queued */ - public int getResponseTimeoutSeconds() { + public long getResponseTimeoutSeconds() { return responseTimeoutSeconds; } @@ -244,7 +244,7 @@ public int getResponseTimeoutSeconds() { * * @param responseTimeoutSeconds - timeout for task to send response. After this timeout, the task will be re-queued */ - public void setResponseTimeoutSeconds(int responseTimeoutSeconds) { + public void setResponseTimeoutSeconds(long responseTimeoutSeconds) { this.responseTimeoutSeconds = responseTimeoutSeconds; } diff --git a/common/src/main/java/com/netflix/conductor/common/run/TaskSummary.java b/common/src/main/java/com/netflix/conductor/common/run/TaskSummary.java index f0fc91361c..a61d13756b 100644 --- a/common/src/main/java/com/netflix/conductor/common/run/TaskSummary.java +++ b/common/src/main/java/com/netflix/conductor/common/run/TaskSummary.java @@ -86,6 +86,9 @@ public class TaskSummary { @ProtoField(id = 16) private String taskId; + public TaskSummary() { + } + public TaskSummary(Task task) { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); diff --git a/common/src/test/java/com/netflix/conductor/common/run/TestTaskSummary.java b/common/src/test/java/com/netflix/conductor/common/run/TestTaskSummary.java new file mode 100644 index 0000000000..5bba3a13c2 --- /dev/null +++ b/common/src/test/java/com/netflix/conductor/common/run/TestTaskSummary.java @@ -0,0 +1,23 @@ +package com.netflix.conductor.common.run; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.netflix.conductor.common.metadata.tasks.Task; +import org.junit.Test; + +import static org.junit.Assert.assertNotNull; + +public class TestTaskSummary { + + @Test + public void testJsonSerializing() throws Exception { + ObjectMapper om = new ObjectMapper(); + + Task task = new Task(); + TaskSummary taskSummary = new TaskSummary(task); + + String json = om.writeValueAsString(taskSummary); + TaskSummary read = om.readValue(json, TaskSummary.class); + assertNotNull(read); + } + +} diff --git a/contribs/src/main/java/com/netflix/conductor/contribs/ContribsModule.java b/contribs/src/main/java/com/netflix/conductor/contribs/ContribsModule.java index e5c1e8b1dc..a9345b68e8 100644 --- a/contribs/src/main/java/com/netflix/conductor/contribs/ContribsModule.java +++ b/contribs/src/main/java/com/netflix/conductor/contribs/ContribsModule.java @@ -58,8 +58,8 @@ protected void configure() { @StringMapKey("sqs") @Singleton @Named(EVENT_QUEUE_PROVIDERS_QUALIFIER) - public EventQueueProvider getSQSEventQueueProvider(AmazonSQSClient amazonSQSClient) { - return new SQSEventQueueProvider(amazonSQSClient); + public EventQueueProvider getSQSEventQueueProvider(AmazonSQSClient amazonSQSClient, Configuration config) { + return new SQSEventQueueProvider(amazonSQSClient, config); } 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 fbedc8d0f9..9a348dc34b 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 @@ -72,7 +72,7 @@ public class SQSObservableQueue implements ObservableQueue { private String queueName; - private int visibilityTimeout; + private int visibilityTimeoutInSeconds; private int batchSize; @@ -82,10 +82,10 @@ public class SQSObservableQueue implements ObservableQueue { private String queueURL; - private SQSObservableQueue(String queueName, AmazonSQSClient client, int visibilityTimeout, int batchSize, int pollTimeInMS, List accountsToAuthorize) { + private SQSObservableQueue(String queueName, AmazonSQSClient client, int visibilityTimeoutInSeconds, int batchSize, int pollTimeInMS, List accountsToAuthorize) { this.queueName = queueName; this.client = client; - this.visibilityTimeout = visibilityTimeout; + this.visibilityTimeoutInSeconds = visibilityTimeoutInSeconds; this.batchSize = batchSize; this.pollTimeInMS = pollTimeInMS; this.queueURL = getOrCreateQueue(); @@ -141,6 +141,18 @@ public String getURI() { return queueURL; } + public int getPollTimeInMS() { + return pollTimeInMS; + } + + public int getBatchSize() { + return batchSize; + } + + public int getVisibilityTimeoutInSeconds() { + return visibilityTimeoutInSeconds; + } + public static class Builder { private String queueName; @@ -271,7 +283,7 @@ List receiveMessages() { try { ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest() .withQueueUrl(queueURL) - .withVisibilityTimeout(visibilityTimeout) + .withVisibilityTimeout(visibilityTimeoutInSeconds) .withMaxNumberOfMessages(batchSize); ReceiveMessageResult result = client.receiveMessage(receiveMessageRequest); diff --git a/contribs/src/main/java/com/netflix/conductor/core/events/sqs/SQSEventQueueProvider.java b/contribs/src/main/java/com/netflix/conductor/core/events/sqs/SQSEventQueueProvider.java index afd4f9772e..77b421e1ba 100644 --- a/contribs/src/main/java/com/netflix/conductor/core/events/sqs/SQSEventQueueProvider.java +++ b/contribs/src/main/java/com/netflix/conductor/core/events/sqs/SQSEventQueueProvider.java @@ -21,6 +21,7 @@ import com.amazonaws.services.sqs.AmazonSQSClient; import com.netflix.conductor.contribs.queue.sqs.SQSObservableQueue; import com.netflix.conductor.contribs.queue.sqs.SQSObservableQueue.Builder; +import com.netflix.conductor.core.config.Configuration; import com.netflix.conductor.core.events.EventQueueProvider; import com.netflix.conductor.core.events.queue.ObservableQueue; @@ -38,21 +39,27 @@ public class SQSEventQueueProvider implements EventQueueProvider { private final Map queues = new ConcurrentHashMap<>(); private final AmazonSQSClient client; + private final int batchSize; + private final int pollTimeInMS; + private final int visibilityTimeoutInSeconds; @Inject - public SQSEventQueueProvider(AmazonSQSClient client) { + public SQSEventQueueProvider(AmazonSQSClient client, Configuration config) { this.client = client; + this.batchSize = config.getIntProperty("workflow.event.queues.sqs.batchSize", 1); + this.pollTimeInMS = config.getIntProperty("workflow.event.queues.sqs.pollTimeInMS", 100); + this.visibilityTimeoutInSeconds = config.getIntProperty("workflow.event.queues.sqs.visibilityTimeoutInSeconds", 60); } @Override public ObservableQueue getQueue(String queueURI) { return queues.computeIfAbsent(queueURI, q -> { Builder builder = new SQSObservableQueue.Builder(); - return builder.withBatchSize(1) + return builder.withBatchSize(this.batchSize) .withClient(client) - .withPollTimeInMS(100) + .withPollTimeInMS(this.pollTimeInMS) .withQueueName(queueURI) - .withVisibilityTimeout(60) + .withVisibilityTimeout(this.visibilityTimeoutInSeconds) .build(); }); } diff --git a/contribs/src/test/java/com/netflix/conductor/core/events/sqs/TestSQSEventQueueProvider.java b/contribs/src/test/java/com/netflix/conductor/core/events/sqs/TestSQSEventQueueProvider.java new file mode 100644 index 0000000000..3763f563cb --- /dev/null +++ b/contribs/src/test/java/com/netflix/conductor/core/events/sqs/TestSQSEventQueueProvider.java @@ -0,0 +1,64 @@ +package com.netflix.conductor.core.events.sqs; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyInt; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.amazonaws.services.sqs.AmazonSQSClient; +import com.amazonaws.services.sqs.model.ListQueuesRequest; +import com.amazonaws.services.sqs.model.ListQueuesResult; +import com.netflix.conductor.contribs.queue.sqs.SQSObservableQueue; +import com.netflix.conductor.core.config.Configuration; +import org.junit.Before; +import org.junit.Test; + +public class TestSQSEventQueueProvider { + private AmazonSQSClient amazonSQSClient; + private Configuration configuration; + + @Before + public void setup() { + amazonSQSClient = mock(AmazonSQSClient.class); + configuration = mock(Configuration.class); + } + + @Test + public void testGetQueueWithDefaultConfiguration() { + when(configuration.getIntProperty(anyString(), anyInt())).thenAnswer(invocation -> invocation.getArguments()[1]); + + ListQueuesResult listQueuesResult = new ListQueuesResult().withQueueUrls("test_queue_1"); + when(amazonSQSClient.listQueues(any(ListQueuesRequest.class))).thenReturn(listQueuesResult); + + SQSEventQueueProvider sqsEventQueueProvider = new SQSEventQueueProvider(amazonSQSClient, configuration); + SQSObservableQueue sqsObservableQueue = (SQSObservableQueue) sqsEventQueueProvider.getQueue("test_queue_1"); + + assertNotNull(sqsObservableQueue); + assertEquals(1, sqsObservableQueue.getBatchSize()); + assertEquals(100, sqsObservableQueue.getPollTimeInMS()); + assertEquals(60, sqsObservableQueue.getVisibilityTimeoutInSeconds()); + } + + @Test + public void testGetQueueWithCustomConfiguration() { + when(configuration.getIntProperty(eq("workflow.event.queues.sqs.batchSize"), anyInt())).thenReturn(10); + when(configuration.getIntProperty(eq("workflow.event.queues.sqs.pollTimeInMS"), anyInt())).thenReturn(50); + when(configuration.getIntProperty(eq("workflow.event.queues.sqs.visibilityTimeoutInSeconds"), anyInt())).thenReturn(30); + + ListQueuesResult listQueuesResult = new ListQueuesResult().withQueueUrls("test_queue_1"); + when(amazonSQSClient.listQueues(any(ListQueuesRequest.class))).thenReturn(listQueuesResult); + + SQSEventQueueProvider sqsEventQueueProvider = new SQSEventQueueProvider(amazonSQSClient, configuration); + SQSObservableQueue sqsObservableQueue = (SQSObservableQueue) sqsEventQueueProvider.getQueue("test_queue_1"); + + assertNotNull(sqsObservableQueue); + assertEquals(10, sqsObservableQueue.getBatchSize()); + assertEquals(50, sqsObservableQueue.getPollTimeInMS()); + assertEquals(30, sqsObservableQueue.getVisibilityTimeoutInSeconds()); + } + +} diff --git a/core/src/main/java/com/netflix/conductor/core/config/CoreModule.java b/core/src/main/java/com/netflix/conductor/core/config/CoreModule.java index 891bd77bdc..f7940e75df 100644 --- a/core/src/main/java/com/netflix/conductor/core/config/CoreModule.java +++ b/core/src/main/java/com/netflix/conductor/core/config/CoreModule.java @@ -144,8 +144,8 @@ public TaskMapper getWaitTaskMapper(ParametersUtils parametersUtils) { @StringMapKey(TASK_TYPE_SUB_WORKFLOW) @Singleton @Named(TASK_MAPPERS_QUALIFIER) - public TaskMapper getSubWorkflowTaskMapper(ParametersUtils parametersUtils) { - return new SubWorkflowTaskMapper(parametersUtils); + public TaskMapper getSubWorkflowTaskMapper(ParametersUtils parametersUtils, MetadataDAO metadataDAO) { + return new SubWorkflowTaskMapper(parametersUtils, metadataDAO); } @ProvidesIntoMap diff --git a/core/src/main/java/com/netflix/conductor/core/events/EventProcessor.java b/core/src/main/java/com/netflix/conductor/core/events/EventProcessor.java index 528380f856..449a238047 100644 --- a/core/src/main/java/com/netflix/conductor/core/events/EventProcessor.java +++ b/core/src/main/java/com/netflix/conductor/core/events/EventProcessor.java @@ -90,6 +90,7 @@ public EventProcessor(ExecutionService executionService, MetadataService metadat executorService = Executors.newFixedThreadPool(executorThreadCount); refresh(); Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(this::refresh, 60, 60, TimeUnit.SECONDS); + logger.info("Event Processing is ENABLED. executorThreadCount set to {}", executorThreadCount); } else { logger.warn("Event processing is DISABLED. executorThreadCount set to {}", executorThreadCount); } @@ -161,6 +162,8 @@ private void handle(ObservableQueue queue, Message msg) { } } catch (Exception e) { logger.error("Error handling message: {} on queue:{}", msg, queue.getName(), e); + } finally { + Monitors.recordEventQueueMessagesHandled(queue.getType(), queue.getName()); } } 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 d7879dce27..91b4af9ca2 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 @@ -448,9 +448,9 @@ void checkForTimeout(TaskDef taskDef, Task task) { return; } - long timeout = 1000 * taskDef.getTimeoutSeconds(); + long timeout = 1000L * taskDef.getTimeoutSeconds(); long now = System.currentTimeMillis(); - long elapsedTime = now - (task.getStartTime() + (task.getStartDelayInSeconds() * 1000)); + long elapsedTime = now - (task.getStartTime() + ((long)task.getStartDelayInSeconds() * 1000L)); if (elapsedTime < timeout) { return; @@ -494,7 +494,7 @@ boolean isResponseTimedOut(TaskDef taskDefinition, Task task) { logger.debug("Evaluating responseTimeOut for Task: {}, with Task Definition: {} ", task, taskDefinition); - long responseTimeout = 1000 * taskDefinition.getResponseTimeoutSeconds(); + long responseTimeout = 1000L * taskDefinition.getResponseTimeoutSeconds(); long now = System.currentTimeMillis(); long noResponseTime = now - task.getUpdateTime(); diff --git a/core/src/main/java/com/netflix/conductor/core/execution/mapper/DecisionTaskMapper.java b/core/src/main/java/com/netflix/conductor/core/execution/mapper/DecisionTaskMapper.java index 3ec5dd1c46..d9f793c769 100644 --- a/core/src/main/java/com/netflix/conductor/core/execution/mapper/DecisionTaskMapper.java +++ b/core/src/main/java/com/netflix/conductor/core/execution/mapper/DecisionTaskMapper.java @@ -84,7 +84,6 @@ public List getMappedTasks(TaskMapperContext taskMapperContext) { decisionTask.setWorkflowType(workflowInstance.getWorkflowName()); decisionTask.setCorrelationId(workflowInstance.getCorrelationId()); decisionTask.setScheduledTime(System.currentTimeMillis()); - decisionTask.setEndTime(System.currentTimeMillis()); decisionTask.getInputData().put("case", caseValue); decisionTask.getOutputData().put("caseOutput", Collections.singletonList(caseValue)); decisionTask.setTaskId(taskId); diff --git a/core/src/main/java/com/netflix/conductor/core/execution/mapper/EventTaskMapper.java b/core/src/main/java/com/netflix/conductor/core/execution/mapper/EventTaskMapper.java index 2a2cd3a808..9cf695a354 100644 --- a/core/src/main/java/com/netflix/conductor/core/execution/mapper/EventTaskMapper.java +++ b/core/src/main/java/com/netflix/conductor/core/execution/mapper/EventTaskMapper.java @@ -61,7 +61,6 @@ public List getMappedTasks(TaskMapperContext taskMapperContext) { eventTask.setWorkflowType(workflowInstance.getWorkflowName()); eventTask.setCorrelationId(workflowInstance.getCorrelationId()); eventTask.setScheduledTime(System.currentTimeMillis()); - eventTask.setEndTime(System.currentTimeMillis()); eventTask.setInputData(eventTaskInput); eventTask.getInputData().put("sink", sink); eventTask.setTaskId(taskId); diff --git a/core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java b/core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java index ea7b7b4b44..433ad7d2d3 100644 --- a/core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java +++ b/core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java @@ -1,4 +1,4 @@ -/** +/* * Copyright 2018 Netflix, Inc. *

* Licensed under the Apache License, Version 2.0 (the "License"); @@ -83,17 +83,16 @@ public ForkJoinDynamicTaskMapper(ParametersUtils parametersUtils, ObjectMapper o *

  • A check is performed that the next following task in the {@link WorkflowDef} is a {@link TaskType#JOIN}
    • * * - * * @param taskMapperContext: A wrapper class containing the {@link WorkflowTask}, {@link WorkflowDef}, {@link Workflow} and a string representation of the TaskId * @throws TerminateWorkflowException In case of: - * + * * @return: List of tasks in the following order: *