[helix] 19/44: TASK: Fix bug in delete()

[hulee]

This is an automated email from the ASF dual-hosted git repository.

hulee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git

commit 8800cb1f16ebe043cf74f6e95caf5d1c9d17c097
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Wed Apr 10 16:56:11 2019 -0700

    TASK: Fix bug in delete()
    
    The delete() call was doing a force delete on workflows created from a 
recurrent workflow. This would cause a race condition between the controller 
cache and the deletion. This diff fixes this.
    Changelist:
    1. Fix the logic in delete()
    
    RB=1627615
    G=helix-reviewers
    A=jxue
    
    Signed-off-by: Hunter Lee <hu...@linkedin.com>
---
 .../src/main/java/org/apache/helix/task/TaskDriver.java  | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java 
b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
index 2f0be85..114b6a2 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
@@ -665,11 +665,17 @@ public class TaskDriver {
     // Delete all previously scheduled workflows.
     if (wCtx != null && wCtx.getScheduledWorkflows() != null) {
       for (String scheduledWorkflow : wCtx.getScheduledWorkflows()) {
-        WorkflowContext scheduledWorkflowCtx =
-            TaskUtil.getWorkflowContext(_propertyStore, scheduledWorkflow);
-        if (scheduledWorkflowCtx != null
-            && scheduledWorkflowCtx.getFinishTime() != 
WorkflowContext.UNFINISHED) {
-          removeWorkflowFromZK(scheduledWorkflow);
+        if (forceDelete) {
+          // Only directly delete it if it's force delete. Otherwise, it will 
cause a race condition
+          // where contexts would get written back to ZK from cache
+          WorkflowContext scheduledWorkflowCtx =
+              TaskUtil.getWorkflowContext(_propertyStore, scheduledWorkflow);
+          if (scheduledWorkflowCtx != null
+              && scheduledWorkflowCtx.getFinishTime() != 
WorkflowContext.UNFINISHED) {
+            removeWorkflowFromZK(scheduledWorkflow);
+          }
+        } else {
+          setWorkflowTargetState(scheduledWorkflow, TargetState.DELETE);
         }
       }
     }