use fallback domain if pollDataDAO is unavailable

Netflix · May 29, 2020 · 6e7d011 · 6e7d011
1 parent b2332d0
commit 6e7d011
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 9 deletions.
diff --git a/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java b/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java
@@ -1245,12 +1245,12 @@ public void executeSystemTask(WorkflowSystemTask systemTask, String taskId, int
     }
 
     @VisibleForTesting
-    void setTaskDomains(List<Task> tasks, Workflow wf) {
-        Map<String, String> taskToDomain = wf.getTaskToDomain();
+    void setTaskDomains(List<Task> tasks, Workflow workflow) {
+        Map<String, String> taskToDomain = workflow.getTaskToDomain();
         if (taskToDomain != null) {
             // Step 1: Apply * mapping to all tasks, if present.
             String domainstr = taskToDomain.get("*");
-            if (domainstr != null) {
+            if (StringUtils.isNotBlank(domainstr)) {
                 String[] domains = domainstr.split(",");
                 tasks.forEach(task -> {
                     // Filter out SystemTask
@@ -1260,7 +1260,6 @@ void setTaskDomains(List<Task> tasks, Workflow wf) {
                         task.setDomain(getActiveDomain(task.getTaskType(), domains));
                     }
                 });
-
             }
             // Step 2: Override additional mappings.
             tasks.forEach(task -> {

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
@@ -25,6 +25,7 @@
 import com.netflix.conductor.core.events.ScriptEvaluator;
 import com.netflix.conductor.core.execution.SystemTaskType;
 import com.netflix.conductor.core.execution.TerminateWorkflowException;
+import org.apache.bval.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -122,7 +123,7 @@ public List<Task> getMappedTasks(TaskMapperContext taskMapperContext) {
     String getEvaluatedCaseValue(WorkflowTask taskToSchedule, Map<String, Object> taskInput) {
         String expression = taskToSchedule.getCaseExpression();
         String caseValue;
-        if (expression != null) {
+        if (StringUtils.isNotBlank(expression)) {
             logger.debug("Case being evaluated using decision expression: {}", expression);
             try {
                 //Evaluate the expression by using the Nashhorn based script evaluator

diff --git a/core/src/main/java/com/netflix/conductor/core/orchestration/ExecutionDAOFacade.java b/core/src/main/java/com/netflix/conductor/core/orchestration/ExecutionDAOFacade.java
@@ -371,7 +371,11 @@ public List<PollData> getTaskPollData(String taskName) {
     }
 
     public PollData getTaskPollDataByDomain(String taskName, String domain) {
-        return pollDataDAO.getPollData(taskName, domain);
+        try {
+            return pollDataDAO.getPollData(taskName, domain);
+        } catch (Exception e) {
+            return null;
+        }
     }
 
     public void updateTaskLastPoll(String taskName, String domain, String workerId) {

diff --git a/core/src/test/java/com/netflix/conductor/core/execution/TestWorkflowExecutor.java b/core/src/test/java/com/netflix/conductor/core/execution/TestWorkflowExecutor.java
@@ -28,7 +28,6 @@
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.booleanThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
@@ -836,6 +835,12 @@ public void testGetActiveDomain() {
 
         activeDomain = workflowExecutor.getActiveDomain(taskType, null);
         assertNull(activeDomain);
+
+        domains = new String[]{"test-domain"};
+        when(executionDAOFacade.getTaskPollDataByDomain(anyString(), anyString())).thenReturn(null);
+        activeDomain = workflowExecutor.getActiveDomain(taskType, domains);
+        assertNotNull(activeDomain);
+        assertEquals("test-domain", activeDomain);
     }
 
     @Test
@@ -859,7 +864,7 @@ public void testDefaultDomain() {
         when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[0])).thenReturn(pollData1);
         when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[1])).thenReturn(null);
         String activeDomain = workflowExecutor.getActiveDomain(taskType, domains);
-        assertEquals(null, activeDomain);
+        assertNull(activeDomain);
     }
 
     @Test

diff --git a/redis-persistence/src/main/java/com/netflix/conductor/dao/dynomite/RedisPollDataDAO.java b/redis-persistence/src/main/java/com/netflix/conductor/dao/dynomite/RedisPollDataDAO.java
@@ -62,7 +62,7 @@ public PollData getPollData(String taskDefName, String domain) {
         recordRedisDaoPayloadSize("getPollData", StringUtils.length(pollDataJsonString), "n/a", "n/a");
 
         PollData pollData = null;
-        if (pollDataJsonString != null) {
+        if (StringUtils.isNotBlank(pollDataJsonString)) {
             pollData = readValue(pollDataJsonString, PollData.class);
         }
         return pollData;