[GitHub] [helix] alirezazamani commented on a change in pull request #1040: Remove previousAssignment in processTaskWithPendingMessage method

[GitBox <gi...@apache.org>]

alirezazamani commented on a change in pull request #1040:
URL: https://github.com/apache/helix/pull/1040#discussion_r433463896



##########
File path: helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java
##########
@@ -449,43 +448,31 @@ private void updatePartitionInformationInJobContext(CurrentStateOutput currentSt
    * @param paMap
    * @param assignedPartitions
    */
-  private void processTaskWithPendingMessage(ResourceAssignment prevAssignment, Integer pId,
-      String pName, String instance, Message pendingMessage, TaskState jobState,
-      TaskPartitionState currState, Map<Integer, PartitionAssignment> paMap,
-      Map<String, Set<Integer>> assignedPartitions) {
-
-    // stateMap is a mapping of Instance -> TaskPartitionState (String)
-    Map<String, String> stateMap = prevAssignment.getReplicaMap(new Partition(pName));
-    if (stateMap != null) {
-      String prevState = stateMap.get(instance);
-      if (!pendingMessage.getToState().equals(prevState)) {
-        LOG.warn(String.format(
-            "Task pending to-state is %s while previous assigned state is %s. This should not"
-                + "happen.",
-            pendingMessage.getToState(), prevState));
+  private void processTaskWithPendingMessage(Integer pId, String pName, String instance,
+      Message pendingMessage, TaskState jobState, TaskPartitionState currState,
+      Map<Integer, PartitionAssignment> paMap, Map<String, Set<Integer>> assignedPartitions) {
+
+    if (jobState == TaskState.TIMING_OUT && currState == TaskPartitionState.INIT
+        && pendingMessage.getToState().equals(TaskPartitionState.RUNNING.name())) {
+      // While job is timing out, if the task is pending on INIT->RUNNING, set it back to INIT,
+      // so that Helix will cancel the transition.
+      paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.INIT.name()));

Review comment:
       @narendly Thank you for the suggestion. I think what you are suggesting would also work. However, I think the logic behind this part of code should be untouched as much as possible. If we want to do send DROPPED here (although I think it would work), we need to release the quota and do further logic changes. The reason is once we DROP the the task from instance, we will miss currentState and we will not see that task in other methods in order to release quota. Hence, I prefer to send DROPPED for the task using two level approach (i.e. first send it to INIT and then DROP it).
   
   Also since there is a pending INIT->RUNNING message existed, it is not very safe to release the quota here. Because the task might go to RUNNING state and before we cancel it, the quota information will be mismatched with the actual scenario. So if we first let the task to go to INIT state and then releasing the quota and sending it to DROPPED will be better option in my opinion.
   
   To answer your question, we will actually handle this case for releasing the quota in updatePreviousAssignedTasksStatus method. Here the code that would do that:
   
   ```
           case INIT: {
             // INIT is a temporary state for tasks
             // Two possible scenarios for INIT:
             // 1. Task is getting scheduled for the first time. In this case, Task's state will go
             // from null->INIT->RUNNING, and this INIT state will be transient and very short-lived
             // 2. Task is getting scheduled for the first time, but in this case, job is timed out or
             // timing out. In this case, it will be sent back to INIT state to be removed. Here we
             // ensure that this task then goes from INIT to DROPPED so that it will be released from
             // AssignableInstance to prevent resource leak
             if (jobState == TaskState.TIMED_OUT || jobState == TaskState.TIMING_OUT
                 || jobTgtState == TargetState.DELETE) {
               // Job is timed out or timing out or targetState is to be deleted, so its tasks will be
               // sent back to INIT
               // In this case, tasks' IdealState will be removed, and they will be sent to DROPPED
               partitionsToDropFromIs.add(pId);
               
               assignedPartitions.get(instance).add(pId);
               paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.DROPPED.name()));
   
               // Also release resources for these tasks
               **assignableInstanceManager.release(instance, taskConfig, quotaType);**
               break;
             } 
   ```
   
   Your comment actually guided me to add another safety check that we can added for INIT case. I added two lines to make sure we DROP the task if job is TIMED_OUT or TIMING_OUT and the task is in INIT state. Thanks for the comments and careful review.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org