populate subworkflow version in mapper

Netflix · Oct 18, 2018 · 582c6cd · 582c6cd
1 parent 2592787
commit 582c6cd
Show file tree

Hide file tree

Showing 20 changed files with 205 additions and 108 deletions.
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
@@ -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
@@ -1,12 +1,12 @@
-/**
+/*
  * Copyright 2016 Netflix, Inc.
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -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<Task> 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
@@ -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"
+  }
+]
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
@@ -1,4 +1,4 @@
-/**
+/*
  * Copyright 2016 Netflix, Inc.
  * <p>
  * 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
@@ -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,15 +236,15 @@ 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;
 	}
 
 	/**
 	 *
 	 * @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/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/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
@@ -84,7 +84,6 @@ public List<Task> 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
@@ -61,7 +61,6 @@ public List<Task> 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);