Merge branch 'dev'

.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/

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;
 
 
 /**

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());
+        }
+    }
 }

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"
+  }
+]

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;
     }
 

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;
 	}
 

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'");

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);
+    }
+
+}

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);
 	}
 
 

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<String> accountsToAuthorize) {
+	private SQSObservableQueue(String queueName, AmazonSQSClient client, int visibilityTimeoutInSeconds, int batchSize, int pollTimeInMS, List<String> 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<Message> receiveMessages() {
 		try {
 			ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest()
 					.withQueueUrl(queueURL)
-					.withVisibilityTimeout(visibilityTimeout)
+					.withVisibilityTimeout(visibilityTimeoutInSeconds)
 					.withMaxNumberOfMessages(batchSize);
 
 			ReceiveMessageResult result = client.receiveMessage(receiveMessageRequest);