Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
In order to support speculative execution for batch jobs(FLIP-168), we need a mechanism to block resources on nodes where the slow tasks are located. We propose to introduce a blocklist mechanism as follows: Once a node is marked as blocked, future slots should not be allocated from the blocked node, but the slots that are already allocated will not be affected.
We introduce a new data structure: BlockedNode to record the information of a blocked node. This information will be recorded in a special component and affect the resource allocation of Flink clusters. However，the blocking of nodes is not permanent, there will be a timeout for it. Once it times out, the nodes will become available again. The overall structure of the blocklist mechanism is shown in the figure below.
In JM, there will be a component (JobMasterBlocklistHandler) responsible for managing all BlockedNode(s) and performing them on SlotPool.
In RM, there will also be a component (ResourceManagerBlocklistHandler) responsible for managing the cluster-level blocklist. SlotManager will be aware of the blocklist information and filter out resources in the blocklist when allocating slots from registered task managers. Similarly, when the ResourceManagerDriver requests new resources from an external resource manager (Yarn or Kubernetes), it also needs to filter out the blocked nodes.
A BlockedNode corresponds to the blocked information of a specific node, including the following 4 fields:
- id: The identifier of the node.
- startTimestamp: The start time of the blocking.
- endTimestamp: The end time of the blocking (at which time the node will become available again).
- cause: The cause for blocking this node.
Blocklist on JobMaster
JobMasterBlocklistHandler is the component in JM responsible for managing all blocked node information and performing them on SlotPool. It consists of two sub-components：BlocklistTracker and BlocklistContext. When receiving a new blocked node information, the JobMasterBlocklistHandler will handle it as following:
- Add the new blocked nodes to the BlocklistTracker.
- Synchronize the new blocked nodes to RM.
- Block the resources on newly added nodes via the BlocklistContext.
BlocklistTracker is the component responsible for tracking blocked nodes. The tracker will regularly remove timeout blocked nodes.
BlocklistContext is the component responsible for blocking slots in SlotPool, the details will be described in SlotPool.
SlotPool should avoid allocating slots that located on blocked nodes. To do that, our core idea is to keep the SlotPool in such a state: there is no slot in SlotPool that is free (no task assigned) and located on blocked nodes. Details are as following:
- When receiving slot offers from task managers located on blocked nodes, all offers should be rejected.
- When a node is newly blocked, we should release all free(no task assigned) slots on it. We need to find all task managers on blocked nodes and release all free slots on them by SlotPoolService#releaseFreeSlotsOnTaskManager.
When a slot state changes from reserved(task assigned) to free(no task assigned), it will check whether the corresponding task manager is blocked. If yes, release the slot.
We will introduce a new slot pool implementation: BlocklistSlotPool, which extends the DefaultDeclarativeSlotPool and overrides some methods to implement the blocklist-related functions described above, and the blocklist information will be passed in the constructor. This new implementation will be only used when the blocklist is enabled.
Blocklist on ResourceManager
ResourceManagerBlocklistHandler is a new component introduced in RM for the blocklist mechanism. It has only one sub-component: BlocklistTracker, which is responsible for managing cluster-level blocklist.
SlotManager should filter out blocked resources when allocating registered resources. To do that, we need following changes:
- When starting SlotManager, the BlockedTaskManagerChecker will be passed in to check whether a registered task manager is located on blocked nodes.
- When trying to fulfill the slot requirements by registered task managers, the task managers located on blocked nodes will be filtered out.
- SlotManager will request new task managers from external resource managers if the registered resources cannot fulfill the requirements. The blocklist also takes effect when requesting new task managers, the details will be described in ResourceManagerDriver.
When deploying Flink on Yarn or Kubernetes, the ResourceManagerDriver is responsible for requesting new task managers from the external resource manager. To avoid allocating task managers from blocked nodes, we need to pass the information of blocked nodes to ResourceManagerDriver at its initialization.
ResourceManagerDriver uses following APIs to tell external resource managers about the information of blocked nodes:
Synchronize Blocklist between JM & RM
The newly added/updated blocked nodes will be synchronized between JM and RM via RPC:
- Once a few blocked nodes are newly added/updated to the JobMasterBlocklistHandler, RM will be notified of these nodes via ResourceManagerGateway#notifyNewBlockedNodes.
- When RM receives the blocked nodes notified by a JM, it will add them into ResourceManagerBlocklistHandler, and notify all JMs of the successfully added/updated nodes through JobMasterGateway#notifyNewBlockedNodes.
- Similarly, when JM receives the blocked nodes notified by RM, it will also add them to JobMasterBlocklistHandler.
Enrich TaskManagerLocation with node information
In order to support blocked nodes, it is necessary to know the node where the task is located. To do that, we need to add a node identifier into TaskManagerLocation. This node identifier should be set by the resource manager when requesting new task managers.
Compatibility, Deprecation, and Migration Plan
The blocklist mechanism is only used when speculative execution is enabled, which entails that Flink's default behavior won't change.
No integration with Flink's web UI
This FLIP does not modify the web UI, but it's in our plan. Currently, we have a preliminary idea of the UI intergration, including improve the slots information displaying and add a page to show the blocklist information.
No support for JM failover
Currently, the blocklist information will be lost after JM failover. In the future, we may persist the blocklist information in HA to support JM failover.
- The changes will be covered by unit and IT cases.
- Test the functionality in a real Standanlone/Yarn/Kubernetes cluster.
No rejected alternatives yet.