[helix] branch master updated: TASK: Drop all tasks whose requested states are DROPPED

[hu...@apache.org]

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


The following commit(s) were added to refs/heads/master by this push:
     new 7259880  TASK: Drop all tasks whose requested states are DROPPED
7259880 is described below

commit 7259880d6bca623ae34468822f41a3d9d086fc4c
Author: Hunter Lee <hu...@linkedin.com>
AuthorDate: Fri Aug 9 16:57:05 2019 -0700

    TASK: Drop all tasks whose requested states are DROPPED
    
    Upon a Participant disconnect, the Participant would carry over from the last session. This would copy all previous task states to the current session and set their requested states as DROPPED (for INIT and RUNNING states).
    
    It came to our attention that sometimes these Participants experience connection issues and the tasks happen to be in TASK_ERROR or COMPLETED states. These tasks would get stuck on the Participant and never be dropped. This issue proposes to add the logic that would get all tasks whose requested states are DROPPED to be dropped immediately.
    Changelist:
    1. Make sure all tasks whose requested state is DROPPED get added to tasksToDrop
    2. Add a unit test: TestDropTerminalTasksUponReset
---
 .../java/org/apache/helix/task/JobDispatcher.java  | 32 +++----
 .../helix/task/TestDropTerminalTasksUponReset.java | 98 ++++++++++++++++++++++
 2 files changed, 116 insertions(+), 14 deletions(-)

diff --git a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
index c8cf09a..c21330b 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
@@ -384,7 +384,7 @@ public class JobDispatcher extends AbstractTaskDispatcher {
    * @param tasksToDrop instance -> pId's, to gather all pIds that need to be dropped
    * @return instance -> partitionIds from previous assignment, if the instance is still live
    */
-  private static Map<String, SortedSet<Integer>> getPrevInstanceToTaskAssignments(
+  protected static Map<String, SortedSet<Integer>> getPrevInstanceToTaskAssignments(
       Iterable<String> liveInstances, ResourceAssignment prevAssignment,
       Set<Integer> allTaskPartitions, CurrentStateOutput currStateOutput, String jobName,
       Map<String, Set<Integer>> tasksToDrop) {
@@ -425,25 +425,29 @@ public class JobDispatcher extends AbstractTaskDispatcher {
         TaskPartitionState currState = TaskPartitionState.valueOf(instanceToCurrState.getValue());
         int pId = TaskUtil.getPartitionId(entry.getKey().getPartitionName());
 
-        // We must add all active task pIds back here because dropping transition could overwrite an
-        // active transition in paMap
-        // Add all task partitions in the following states:
-        // currState = INIT, requestedState = RUNNING (bootstrap)
-        // currState = RUNNING, requestedState = ANY (active)
-        // ** for tasks that are just in INIT state, we do not add them here because old
-        // Participants, upon connection reset, set tasks' currentStates to INIT. We cannot consider
-        // those tasks active **
-        if (result.containsKey(instance) && (currState == TaskPartitionState.INIT
-            && requestedState != null && requestedState.equals(TaskPartitionState.RUNNING.name())
-            || currState == TaskPartitionState.RUNNING)) {
-          // Check if this is a dropping transition
+        if (result.containsKey(instance)) {
+          // We must add all active task pIds back here because dropping transition could overwrite
+          // an active transition in paMap
+          // Add all task partitions in the following states:
+          // currState = INIT, requestedState = RUNNING (bootstrap)
+          // currState = RUNNING, requestedState = ANY (active)
+          // ** for tasks that are just in INIT state, we do not add them here because old
+          // Participants, upon connection reset, set tasks' currentStates to INIT. We cannot
+          // consider those tasks active **
+          if (currState == TaskPartitionState.INIT && requestedState != null
+              && requestedState.equals(TaskPartitionState.RUNNING.name())
+              || currState == TaskPartitionState.RUNNING) {
+            result.get(instance).add(pId);
+          }
+
+          // Check if this task needs to be dropped. If so, we need to add to tasksToDrop no matter
+          // what its current state is so that it will be dropped
           if (requestedState != null && requestedState.equals(TaskPartitionState.DROPPED.name())) {
             if (!tasksToDrop.containsKey(instance)) {
               tasksToDrop.put(instance, new HashSet<>());
             }
             tasksToDrop.get(instance).add(pId);
           }
-          result.get(instance).add(pId);
         }
       }
     }
diff --git a/helix-core/src/test/java/org/apache/helix/task/TestDropTerminalTasksUponReset.java b/helix-core/src/test/java/org/apache/helix/task/TestDropTerminalTasksUponReset.java
new file mode 100644
index 0000000..eff51d1
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/task/TestDropTerminalTasksUponReset.java
@@ -0,0 +1,98 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ * 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.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import org.apache.helix.controller.stages.CurrentStateOutput;
+import org.apache.helix.model.Partition;
+import org.apache.helix.model.ResourceAssignment;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+public class TestDropTerminalTasksUponReset {
+
+  /**
+   * This is a unit test that tests that all task partitions with requested state = DROPPED will be
+   * added to tasksToDrop.
+   */
+  @Test
+  public void testDropAllTerminalTasksUponReset() {
+    Random random = new Random();
+    String jobName = "job";
+    String nodeName = "localhost";
+    int numTasks = 10;
+
+    // Create an Iterable of LiveInstances
+    Collection<String> liveInstances = new HashSet<>();
+    liveInstances.add("localhost");
+
+    // Create a dummy ResourceAssignment
+    ResourceAssignment prevAssignment = new ResourceAssignment(jobName);
+
+    // Create allTaskPartitions
+    Set<Integer> allTaskPartitions = new HashSet<>();
+
+    // Create a mock CurrentStateOutput
+    CurrentStateOutput currentStateOutput = mock(CurrentStateOutput.class);
+
+    // Generate a CurrentStateMap
+    Map<Partition, Map<String, String>> currentStateMap = new HashMap<>();
+    when(currentStateOutput.getCurrentStateMap(jobName)).thenReturn(currentStateMap);
+
+    for (int i = 0; i < numTasks; i++) {
+      allTaskPartitions.add(i);
+      Partition task = new Partition(jobName + "_" + i);
+      currentStateMap.put(task, new HashMap<>());
+
+      // Pick some random currentState between COMPLETED and TASK_ERROR
+      String currentState = (random.nextBoolean()) ? TaskPartitionState.COMPLETED.name()
+          : TaskPartitionState.TASK_ERROR.name();
+
+      // First half of the tasks to be dropped on each instance
+      if (i < numTasks / 2) {
+        // requested state is DROPPED
+        currentStateMap.get(task).put("localhost", currentState);
+        when(currentStateOutput.getRequestedState(jobName, task, nodeName))
+            .thenReturn(TaskPartitionState.DROPPED.name());
+      } else {
+        // requested state is nothing
+        when(currentStateOutput.getRequestedState(jobName, task, nodeName)).thenReturn(null);
+      }
+    }
+
+    // Create an empty tasksToDrop
+    Map<String, Set<Integer>> tasksToDrop = new HashMap<>();
+
+    // Call the static method we are testing
+    JobDispatcher.getPrevInstanceToTaskAssignments(liveInstances, prevAssignment, allTaskPartitions,
+        currentStateOutput, jobName, tasksToDrop);
+
+    // Check that tasksToDrop has (numTasks / 2) partitions as we intended regardless of what the
+    // current states of the tasks were
+    Assert.assertEquals(numTasks / 2, tasksToDrop.get(nodeName).size());
+  }
+}