[jira] [Created] (FLINK-14589) Redundant slot requests with the same AllocationID leads to inconsistent slot table

[Shang Yuanchun (Jira)]

Hwanju Kim created FLINK-14589:
----------------------------------

             Summary: Redundant slot requests with the same AllocationID leads to inconsistent slot table
                 Key: FLINK-14589
                 URL: https://issues.apache.org/jira/browse/FLINK-14589
             Project: Flink
          Issue Type: Bug
          Components: Runtime / Coordination
    Affects Versions: 1.6.3
            Reporter: Hwanju Kim


_NOTE: We found this issue in 1.6.2, but I checked the relevant code is still in the mainline. What I am not sure, however, is what other slot-related fixes after 1.6.2 (such as FLINK-11059 and FLINK-12863, etc) would prevent the initial cause of this issue from happening. So far, I have not found the related fix to the issue I am describing here, so opening this issue. Please feel free to deduplicate this if another one already covers it. Please note that we have already picked FLINK-9912, which turned out to be a major fix to slot allocation failure issue. I will note the ramification to that issue just in case others experience the same problem)._
h2. Summary

When *requestSlot* is called from ResourceManager (RM) to TaskManager (TM), TM firstly reserves the requested slot marking it as ALLOCATED, offers the slot to JM, and marks the slot as ACTIVE once getting acknowledgement from JM. This three-way communication for slot allocation is identified by AllocationID, which is generated by JM initially. The way TM reserves a slot is by calling *TaskSlotTable.allocateSlot* if the requested slot number (i.e., slot index) is free to use. The major data structure is *TaskSlot* indexed by slot index. Once the slot is marked as ALLOCATED with a given AllocationID, it tries to update other maps such as *allocationIDTaskSlotMap* keyed by AllocationID and *slotsPerJob* keyed by JobID. Here when updating *allocationIDTaskSlotMap*, it's directly using *allocationIDTaskSlotMap.put(allocationId, taskSlot)*, which may overwrite existing entry, if one is already there with the same AllocationID. This would render inconsistency between *TaskSlot* and *allocationIDTaskSlotMap*, where the former says two slots are allocated by the same AllocationID and the latter says the AllocationID only has the latest task slot. With this state, once the slot is freed, *freeSlot* is driven by AllocationID, so it fetches slot index (i.e., the latter one that has arrived later) from *allocationIDTaskSlotMap*, marks the slot free, and removes it from *allocationIDTaskSlotMap*. But still the old task slot is marked as allocated. This old task slot becomes zombie and can never be freed. This can cause permanent slot allocation failure if TM slots are statically and tightly provisioned and resource manager is not actively spawning new TMs where unavailable (e.g., Kubernetes without active mode integration, which is not yet available).
h2. Scenario

From my observation, the redundant slot requests with the same AllocationID and different slot indices should be rare but can happen with race condition especially when repeated fail-over and heartbeat timeout (primarily caused by transient resource overload, not permanent network partition/node outage) are taking place. The following is a detailed scenario, which could lead to this issue (AID is AllocationID):
 # AID1 is requested from JM and put in the pending request queue in RM.
 # RM picks up slot number 1 (Slot1) from freeSlots and performs requestSlot with Slot1 and AID1. Here this slot request is on the fly.
 # In the meantime, Slot1 is occupied by AID2 in TM for a delayed slot request and TM sends slot report via heartbeat to RM saying Slot1 is already allocated with AID2.
 # RM's heartbeat handler identifies that Slot1 is occupied with a different AID (AID2) so that it should reject the pending request sent from step 2.
 # handleFailedSlotRequest puts the rejected AID1 to pending request again by retrying the slot request. Now it picks up another available slot, say Slot2. So, the retried slot request with Slot 2 and AID1 is on the fly.
 # In the meantime, Slot1 occupied by AID2 is freed (by any disconnection with JM, or releasing all the tasks in the slot on cancellation/failure - the latter was observed).
 # The in-flight slot request (Slot1, AID1) from step 2 arrives at TM, and it's succeeded as Slot1 is free to allocate. TM offers the Slot1 to JM, which acknowledges it so that TM marks Slot1 ACTIVE with AID1. As this point, allocationIDTaskSlotMap[AID1] = Slot1 in TM. JM's allocatedSlots[AID1] = Slot1.
 # The next in-flight slot request (Slot2, AID1) from step 5 arrives at TM. As Slot2 is still free, TM marks it ALLOCATED and offers Slot2 to JM and *"overwrite allocationIDTaskSlotMap[AID1] to Slot2"*
 # In step 7, JM has allocatedSlots[AID1] = Slot1, which leads JM to reject the offer as the same AID is already occupied by another slot.
 # TM gets the rejected offer for (Slot2, AID1) and frees Slot2. As part of that, it removes allocationIDTaskSlotMap[AID1]. *Here Slot1 is still marked as ALLOCATED with AID1 but allocationIDTaskSlotMap contains nothing for AID1.*
 # From this point on, RM believes that Slot1 is allocated for AID1, so is JM, proceeding task deployment with AID1. In TM, AID1 is not allocated at all due to allocationIDTaskSlotMap[AID1] = null. Task deployment is failed with *TaskSubmissionException("No task slot allocated for job ID")*.
 # Any slot release from JM (by another heartbeat timeout) removes the allocated slot (Slot1, AID1) from allocatedSlots and availableSlots, where freeing slots with AID1 in TM is no-op due to allocationIDTaskSlotMap[AID1] = null.
 # Any further scheduling is failed with *NoResourceAvailableException("Could not allocate all requires slots within timeout")*, unless active resource manager is used.

h2. Repro method

The repro I have is a little stressed way than deterministic one, by having constantly failing app (also with not reacting to cancellation to prolong fail-over), with short heartbeat timeout (<=1s) to emulate resource overload without imposing arbitrary resource overloads (I enabled DEBUG log, which could also add a bit more load). Apart from repro test, the real applications that run into this issue are heavily loaded (high slot oversubscription) doing continuous fail-overs (by code error).
h2. Experiment

I've tested the simple solution like having *TaskSlotTable.allocateSlot* check if *allocationIDTaskSlotMap* has an existing task slot entry for a requested AID, it can reject allocation up front, so that the redundant slot request can fail fast not reaching reservation and slot offer, preventing allocationIDTaskSlotMap overwrite and inconsistent allocation tables. In the above example, the fix can replace step 8, 9, and 10 with RM getting *SlotAllocationException* on handling failed slot request. It may retry the pending request via handleFailedSlotRequest, but it would stop retrying once next heartbeat saying the pending slot was allocated by another AID or eventually slot request timeout. My test with the fix has run for a while hitting this race case multiple times and survived without slot allocation failure.

As my experience may be in a bit old version (1.6.2), It would be good to discuss what other unexpected things happened and what would be other related issues that have been incorporated in mainline.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)