remove obsolete TODOs and refactor (Netflix#3161)

* remove obsolete TODOs and refactor * spotless * using mockbean * Revert "using mockbean" This reverts commit e1cb986. * revert changes to event processor * fix tests
routific · Aug 10, 2022 · 290ab2b · 290ab2b
1 parent 7738287
commit 290ab2b
Show file tree

Hide file tree

Showing 32 changed files with 42,806 additions and 3,593 deletions.
diff --git a/annotations-processor/dependencies.lock b/annotations-processor/dependencies.lock
diff --git a/annotations/dependencies.lock b/annotations/dependencies.lock
diff --git a/awss3-storage/dependencies.lock b/awss3-storage/dependencies.lock
diff --git a/awssqs-event-queue/dependencies.lock b/awssqs-event-queue/dependencies.lock
diff --git a/cassandra-persistence/dependencies.lock b/cassandra-persistence/dependencies.lock
diff --git a/client-spring/dependencies.lock b/client-spring/dependencies.lock
diff --git a/client/dependencies.lock b/client/dependencies.lock
diff --git a/common/dependencies.lock b/common/dependencies.lock
diff --git a/core/dependencies.lock b/core/dependencies.lock
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
@@ -90,6 +90,7 @@ public DefaultEventProcessor(
         this.objectMapper = objectMapper;
         this.jsonUtils = jsonUtils;
         this.evaluators = evaluators;
+        this.retryTemplate = retryTemplate;
 
         if (properties.getEventProcessorThreadCount() <= 0) {
             throw new IllegalStateException(
@@ -105,7 +106,6 @@ public DefaultEventProcessor(
                         properties.getEventProcessorThreadCount(), threadFactory);
 
         this.isEventMessageIndexingEnabled = properties.isEventMessageIndexingEnabled();
-        this.retryTemplate = retryTemplate;
         LOGGER.info("Event Processing is ENABLED");
     }
 

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
@@ -12,11 +12,7 @@
  */
 package com.netflix.conductor.core.events;
 
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang3.StringUtils;

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
@@ -838,11 +838,6 @@ WorkflowModel completeWorkflow(WorkflowModel workflow) {
             throw new ConflictException(msg);
         }
 
-        // FIXME Backwards compatibility for legacy workflows already running.
-        // This code will be removed in a future version.
-        if (workflow.getWorkflowDefinition() == null) {
-            workflow = metadataMapperService.populateWorkflowWithDefinitions(workflow);
-        }
         deciderService.updateWorkflowOutput(workflow, null);
 
         workflow.setStatus(WorkflowModel.Status.COMPLETED);
@@ -906,12 +901,6 @@ public WorkflowModel terminateWorkflow(
                 workflow.setStatus(WorkflowModel.Status.TERMINATED);
             }
 
-            // FIXME Backwards compatibility for legacy workflows already running.
-            // This code will be removed in a future version.
-            if (workflow.getWorkflowDefinition() == null) {
-                workflow = metadataMapperService.populateWorkflowWithDefinitions(workflow);
-            }
-
             try {
                 deciderService.updateWorkflowOutput(workflow, null);
             } catch (Exception e) {
@@ -1037,13 +1026,6 @@ public void updateTask(TaskResult taskResult) {
         String workflowId = taskResult.getWorkflowInstanceId();
         WorkflowModel workflowInstance = executionDAOFacade.getWorkflowModel(workflowId, false);
 
-        // FIXME Backwards compatibility for legacy workflows already running.
-        // This code will be removed in a future version.
-        if (workflowInstance.getWorkflowDefinition() == null) {
-            workflowInstance =
-                    metadataMapperService.populateWorkflowWithDefinitions(workflowInstance);
-        }
-
         TaskModel task =
                 Optional.ofNullable(executionDAOFacade.getTaskModel(taskResult.getTaskId()))
                         .orElseThrow(
@@ -1288,10 +1270,6 @@ public boolean decide(String workflowId) {
         // If it is a new workflow, the tasks will be still empty even though include tasks is true
         WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
 
-        // FIXME Backwards compatibility for legacy workflows already running.
-        // This code will be removed in a future version.
-        workflow = metadataMapperService.populateWorkflowWithDefinitions(workflow);
-
         if (workflow.getStatus().isTerminal()) {
             if (!workflow.getStatus().isSuccessful()) {
                 cancelNonTerminalTasks(workflow);
@@ -1539,10 +1517,6 @@ public void skipTaskFromWorkflow(
 
         WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
 
-        // FIXME Backwards compatibility for legacy workflows already running.
-        // This code will be removed in a future version.
-        workflow = metadataMapperService.populateWorkflowWithDefinitions(workflow);
-
         // If the workflow is not running then cannot skip any task
         if (!workflow.getStatus().equals(WorkflowModel.Status.RUNNING)) {
             String errorMsg =

diff --git a/core/src/main/java/com/netflix/conductor/service/ExecutionService.java b/core/src/main/java/com/netflix/conductor/service/ExecutionService.java
@@ -12,33 +12,18 @@
  */
 package com.netflix.conductor.service;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.stream.Collectors;
 
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import com.netflix.conductor.annotations.Trace;
 import com.netflix.conductor.common.metadata.events.EventExecution;
-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.run.ExternalStorageLocation;
-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.common.metadata.tasks.*;
+import com.netflix.conductor.common.run.*;
 import com.netflix.conductor.common.utils.ExternalPayloadStorage;
 import com.netflix.conductor.common.utils.ExternalPayloadStorage.Operation;
 import com.netflix.conductor.common.utils.ExternalPayloadStorage.PayloadType;
@@ -612,13 +597,25 @@ public List<TaskExecLog> getTaskLogs(String taskId) {
     /**
      * Get external uri for the payload
      *
-     * @param operation the type of {@link Operation} to be performed
-     * @param payloadType the {@link PayloadType} at the external uri
      * @param path the path for which the external storage location is to be populated
+     * @param operation the type of {@link Operation} to be performed
+     * @param type the {@link PayloadType} at the external uri
      * @return the external uri at which the payload is stored/to be stored
      */
     public ExternalStorageLocation getExternalStorageLocation(
-            Operation operation, PayloadType payloadType, String path) {
-        return externalPayloadStorage.getLocation(operation, payloadType, path);
+            String path, String operation, String type) {
+        try {
+            ExternalPayloadStorage.Operation payloadOperation =
+                    ExternalPayloadStorage.Operation.valueOf(StringUtils.upperCase(operation));
+            ExternalPayloadStorage.PayloadType payloadType =
+                    ExternalPayloadStorage.PayloadType.valueOf(StringUtils.upperCase(type));
+            return externalPayloadStorage.getLocation(payloadOperation, payloadType, path);
+        } catch (Exception e) {
+            String errorMsg =
+                    String.format(
+                            "Invalid input - Operation: %s, PayloadType: %s", operation, type);
+            LOGGER.error(errorMsg);
+            throw new IllegalArgumentException(errorMsg);
+        }
     }
 }
diff --git a/core/src/main/java/com/netflix/conductor/service/TaskServiceImpl.java b/core/src/main/java/com/netflix/conductor/service/TaskServiceImpl.java
@@ -33,7 +33,6 @@
 import com.netflix.conductor.common.run.ExternalStorageLocation;
 import com.netflix.conductor.common.run.SearchResult;
 import com.netflix.conductor.common.run.TaskSummary;
-import com.netflix.conductor.common.utils.ExternalPayloadStorage;
 import com.netflix.conductor.core.utils.QueueUtils;
 import com.netflix.conductor.dao.QueueDAO;
 import com.netflix.conductor.metrics.Monitors;
@@ -365,22 +364,6 @@ public SearchResult<Task> searchV2(
      */
     public ExternalStorageLocation getExternalStorageLocation(
             String path, String operation, String type) {
-        try {
-            ExternalPayloadStorage.Operation payloadOperation =
-                    ExternalPayloadStorage.Operation.valueOf(StringUtils.upperCase(operation));
-            ExternalPayloadStorage.PayloadType payloadType =
-                    ExternalPayloadStorage.PayloadType.valueOf(StringUtils.upperCase(type));
-            return executionService.getExternalStorageLocation(payloadOperation, payloadType, path);
-        } catch (Exception e) {
-            // FIXME: for backwards compatibility
-            LOGGER.error(
-                    "Invalid input - Operation: {}, PayloadType: {}, defaulting to WRITE/TASK_OUTPUT",
-                    operation,
-                    type);
-            return executionService.getExternalStorageLocation(
-                    ExternalPayloadStorage.Operation.WRITE,
-                    ExternalPayloadStorage.PayloadType.TASK_OUTPUT,
-                    path);
-        }
+        return executionService.getExternalStorageLocation(path, operation, type);
     }
 }
diff --git a/core/src/main/java/com/netflix/conductor/service/WorkflowServiceImpl.java b/core/src/main/java/com/netflix/conductor/service/WorkflowServiceImpl.java
@@ -17,7 +17,6 @@
 import java.util.Map;
 import java.util.Optional;
 
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
@@ -32,7 +31,6 @@
 import com.netflix.conductor.common.run.SearchResult;
 import com.netflix.conductor.common.run.Workflow;
 import com.netflix.conductor.common.run.WorkflowSummary;
-import com.netflix.conductor.common.utils.ExternalPayloadStorage;
 import com.netflix.conductor.core.exception.NotFoundException;
 import com.netflix.conductor.core.execution.WorkflowExecutor;
 import com.netflix.conductor.core.utils.Utils;
@@ -520,22 +518,6 @@ public SearchResult<Workflow> searchWorkflowsByTasksV2(
      */
     public ExternalStorageLocation getExternalStorageLocation(
             String path, String operation, String type) {
-        try {
-            ExternalPayloadStorage.Operation payloadOperation =
-                    ExternalPayloadStorage.Operation.valueOf(StringUtils.upperCase(operation));
-            ExternalPayloadStorage.PayloadType payloadType =
-                    ExternalPayloadStorage.PayloadType.valueOf(StringUtils.upperCase(type));
-            return executionService.getExternalStorageLocation(payloadOperation, payloadType, path);
-        } catch (Exception e) {
-            // FIXME: for backwards compatibility
-            LOGGER.error(
-                    "Invalid input - Operation: {}, PayloadType: {}, defaulting to WRITE/WORKFLOW_INPUT",
-                    operation,
-                    type);
-            return executionService.getExternalStorageLocation(
-                    ExternalPayloadStorage.Operation.WRITE,
-                    ExternalPayloadStorage.PayloadType.WORKFLOW_INPUT,
-                    path);
-        }
+        return executionService.getExternalStorageLocation(path, operation, type);
     }
 }
diff --git a/core/src/test/java/com/netflix/conductor/core/events/TestSimpleActionProcessor.java b/core/src/test/java/com/netflix/conductor/core/events/TestSimpleActionProcessor.java
@@ -40,18 +40,9 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyMap;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
 
 @ContextConfiguration(classes = {TestObjectMapperConfiguration.class})
 @RunWith(SpringRunner.class)