[Code] conductor-common

[kymr]

Common models used by various conductor modules

[kymr]

utils

• RetryUtil.java
  • RetryUtil implementation based on guava retrying
  • The WaitStrategies already set to default exponential and random

.withWaitStrategy(WaitStrategies.join(
                        WaitStrategies.exponentialWait(1000, 90, TimeUnit.SECONDS),
                        WaitStrategies.randomWait(100, TimeUnit.MILLISECONDS, 500, TimeUnit.MILLISECONDS)
                ))

[kymr]

run

• SearchResult.java

  • SearchResult Generic

• TaskSummary.java

  • TimeZone is GMT
  • It represents summarized task information.

• Workflow.java

  • WorkflowStatus - enum
  • values
    • RUNNING
    • COMPLETED
    • FAILED
    • TIMED_OUT
    • TERMINATED
    • PAUSED
  • terminal / successful
  • It represents workflow information.
  • default status is RUNNING

• WorkflowSummary.java

  • TimeZone is GMT
  • It represents summarized workflow information.
  • This class is not for Definition

[kymr]

metadata

• Auditable.java

events

• EventExecution.java
  • information object about event-execution
• EventHandler.java
  • It has a list of actions
  • Action
  • Type
    • start_workflow
    • complete_task
    • fail_task
  • TaskDetails
    • this is the result of whether it is succeeded or failed
  • StartWorkflow
    • information when action's type is start_workflow

tasks

• PollData.java

• Task.java

  • Status

    IN_PROGRESS(false, true, true),
    CANCELED(true, false, false),
    FAILED(true, false, true),
    FAILED_WITH_TERMINAL_ERROR(true, false, false), //No Retires even if retries are configured, the task and the related workflow should be terminated
    COMPLETED(true, true, true),
    COMPLETED_WITH_ERRORS(true, true, true),
    SCHEDULED(false, true, true),
    TIMED_OUT(true, false, true),
    READY_FOR_RERUN(false, true, true),
    SKIPPED(true, true, false);
    
    private boolean terminal;
    private boolean successful;
    private boolean retriable;

• TaskDef.java
  • TimeoutPolicy
  • RetryLogic
  • concurrentExecLimit
  • concurrent consumer control
• TaskExecLog.java
• TaskResult.java
  • callbackAfterSeconds
  • for long running task, this is delayed queue implementation
  • acknowledge implementation

workflow

• DynamicForkJoinTask.java
• DynamicForkJoinTaskList.java
• RerunWorkflowRequest.java
• SkipTaskRequest.java
• StartWorkflowRequest.java
• SubWorkflowParams.java

• WorkflowDef.java

  • LinkedList
  • failureWorkflow
  • The part of the workflow order is implemented in both directions, not in on direction

  • see carefully getNextTask

    public WorkflowTask getNextTask(String taskReferenceName){
    Iterator<WorkflowTask> it = tasks.iterator();
    while(it.hasNext()){
         WorkflowTask task = it.next();
         WorkflowTask nextTask = task.next(taskReferenceName, null);
         if(nextTask != null){
             return nextTask;
         }
    
         if(task.getTaskReferenceName().equals(taskReferenceName) || task.has(taskReferenceName)){
             break;
         }
    }
    if(it.hasNext()){
        return it.next();
    }
    return null;
    }

  • Because workflowTask could be sub-worflow or it could be dynamic, it seems to have implemented that way.

• WorkflowTask.java

  • inputParameters

    //Key: Name of the input parameter.  MUST be one of the keys defined in TaskDef (e.g. fileName)
    //Value: mapping of the parameter from another task (e.g. task1.someOutputParameterAsFileName)
    private Map<String, Object> inputParameters = new HashMap<String, Object>();

  • children method returns a list of list of workflowTasks, when type is DECISION or FORK_JOIN

    
    private Collection<List<WorkflowTask>> children(){
    Collection<List<WorkflowTask>> v1 = new LinkedList<>();
    Type tt = Type.USER_DEFINED;
    if(Type.isSystemTask(type)) {
        tt = Type.valueOf(type);
    }
    
    switch(tt){
        case DECISION:
            v1.addAll(decisionCases.values());
            v1.add(defaultCase);
            break;
        case FORK_JOIN:
            v1.addAll(forkTasks);
            break;
        default:
            break;
    }
    return v1;

  }

  
  * This traversal was so confused by taskReferenceName. It is the current taskReferenceName that is the basis to look for next

[kymr]

CopyOnWriteArrayList

• https://docs.oracle.com/javase/8/docs/api/?java/util/concurrent/CopyOnWriteArrayList.html
• A thread-safe variant of ArrayList in which all mutative operations (add, set, and so on) are implemented by making a fresh copy of the underlying array.
• useful when you have few writes and many reads