Status

Current state: Draft

Discussion thread:  None

JIRA: None

Released: None

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Flink jobs typically run in a distributed way. In large clusters, it’s very common for cluster nodes to encounter issues, such as insufficient disk space, bad hardware, network abnormalities. These problems will cause tasks to run abnormally (failure or run slowly). The JM will take care of that and redeploy the relevant tasks. However, due to data locality and limited resources, the new tasks are very likely to be redeployed to the same nodes, which will result in continuous task abnormalities and affect job progress. 

We propose to introduce the blacklist mechanism to solve this problem. Blacklist is a mechanism to filter out problematic resources. Once a resource is judged to be abnormal, it will be blacklisted to avoid assigning tasks to it. We will introduce following two ways to blacklist resources:

1. Manually specify the blacklisted resources through REST API. When users find abnormal nodes/TMs, they can manually blacklist them in this way.
2. Automatically detect abnormal resources and blacklist them. Users can specify a blacklist strategy which identifies abnormal resources according to the received exception and related locations.

Public Interfaces

We propose to introduce following configuration options for blacklist:

• cluster.resource-blacklist.enabled: Whether to enable blacklist mechanism.
• cluster.resource-blacklist.item.timeout: Timeout for blacklisted items in blacklist to be removed
• cluster.resource-blacklist.item.timeout-check-interval: The check interval for blacklist items timeout.

Proposed Changes

In this design, two granularities of blacklisted resources are supported: task managers and nodes. A record of blacklist information is called a blacklisted item, which is generally generated by the scheduler according to the exception of the tasks. These blacklisted items will be recorded in a special component and affect the resource allocation of Flink clusters. However，the blacklist items are not permanent, there will be a timeout for it. Once an item times out, it will be removed, and the resource will become available again. The overall structure of the blacklist mechanism is shown in the figure below. 

In JM, there will be a component (JobMasterBlacklistHandler) responsible for generating blacklisted items according to the exceptions notified by Scheduler, and managing them. SlotPool will be aware of the blacklist information and filter out blacklisted resources when allocating slots.

In RM, there will also be a component (ResourceManagerBlacklistHandler) responsible for managing the cluster-level blacklist. SlotManager will be aware of the blacklist information  and filter out resources in the blacklist 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 blacklisted nodes.

Blacklisted Item

A blacklisted item corresponds to the blacklist information of a specific task manager or node, including the following 5 fields:

1. type: The blacklisted item type, TASK_MANAGER or NODE
2. timestamp: The timestamp for creating this item, will be used to check timeout.
3. cause: The cause for creating this item.
4. identifier: The identifier of the blacklisted task manager or node.
5. action: The action when a task manager/node is marked as blacklisted, including:
   1. MARK_BLACKLISTED: Just mark the task manager or node as blacklisted. Future slots should not be allocated from the blacklisted task manager or node. But slots that are already allocated will not be affected.
   2. MARK_BLACKLISTED_AND_EVACUATE_TASKS: Mark the task manager or node as blacklisted, and evacuate all tasks on it. Evacuated tasks will be restarted on non-blacklisted task managers.

/**
 * This class represents a blacklisted item.
 *
 * @param <ID> Identifier of the blacklisted item.
 */
public abstract class BlacklistedItem<ID> {
    public BlacklistedItemType getType();

    public long getTimestamp();

    public BlacklistAction getAction();

    public Throwable getCause();

    public abstract ID getIdentifier();
}

/** This class represents a blacklisted node. */
public class BlacklistedNode extends BlacklistedItem<String> {
}

/** This class represents a blacklisted task manager. */
public class BlacklistedTaskManager extends BlacklistedItem<ResourceID> {
}

Blacklist on JobMaster

JobMasterBlacklistHandler

JobMasterBlacklistHandler is the component in JM responsible for generating and managing blacklisted items. It consists of three sub-components：BlacklistStrategy, BlacklistTracker and BlacklistContext. When receiving a new exception from Scheduler, the JobMasterBlacklistHandler will handle it as following:

1. Generate new blacklisted items by notifying the exception to the BlacklistStrategy.
2. Add the new blacklisted items to the BlacklistTracker. 
3. Synchronize the new blacklisted items to RM.
4. Perform blacklist actions on the resources via the BlacklistContext.

BlacklistStrategy

BlacklistStrategy is the component responsible for generating blacklist items according to the exceptions and their locations notified by Scheduler. We can introduce different BlacklistStrategy implementations to adapt to different scenarios.

public interface BlacklistStrategy {
    /**
     * Generate blacklisted items according to the abnormal task's location and cause.
     *
     * @param locations the abnormal tasks’ locations.
     * @param cause the cause of blacklisted items.
     * @param timestamp the create timestamp of blacklisted items.
     * @return the generated blacklisted items.
     */
    Collection<BlacklistedItem<?>> generateBlacklistedItems(Collection<TaskManagerLocation> locations, Throwable cause, long timestamp);
}

BlacklistTracker

BlacklistTracker is the component responsible for tracking blacklist items. The tracker will regularly remove timeout blacklisted items.

public interface BlacklistTracker {
    /** Starts the blacklist tracker. */
    void start(ComponentMainThreadExecutor mainThreadExecutor);

    /**
     * Add new blacklisted items or update existing items.
     *
     * @param items The items to add or update
     * @return Newly added or updated items.
     */
    Collection<BlacklistedItem<?>> addNewBlacklistedItems(Collection<BlacklistedItem<?>> items);

    /** Returns whether the given task manager is blacklisted. */
    boolean isBlacklistedTaskManager(ResourceID resourceID);

    /** Get all blacklisted nodes. */
    Set<String> getBlacklistedNodes();

    /** Close the blacklist tracker. */
    void close();
}
     

BlacklistContext

BlacklistContext is the component responsible for performing blacklist actions on SlotPool, the details will be described in SlotPool.

public interface BlacklistContext {
    /** Perform the newly added or updated blacklist items on resources. */
    void blacklistResources(Collection<BlacklistedItem<?>> newlyAddedOrUpdatedItems);
}
   

public interface BlacklistHandler extends BlacklistTracker {
}

public interface JobMasterBlacklistHandler extends BlacklistHandler {

    /**
     * Notify an exception that may generate blacklist items.
     *
     * @param locations locations of the exception
     * @param cause the exception
     */
    void notifyException(Collection<TaskManagerLocation> locations, Throwable cause);
}

SlotPool

SlotPool should avoid allocating slots from blacklisted task managers. Blacklisted task managers include those directly blacklisted and those located on blacklisted 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 blacklisted. Details are as following:

1. When SlotPool starts, BlacklistedTaskManagerChecker will be passed in to check whether a task manager is blacklisted.
2. When receiving slot offers from blacklisted task managers (including task managers on blacklisted nodes), all offers should be rejected.
3. When a task manager is newly blacklisted, BlacklistContext will perform the following on SlotPool:
   1. If the action is MARK_BLACKLISTED, release all free(no task assigned) slots on the blacklisted task manager.
   2. If the action is MARK_BLACKLISTED_AND_EVACUATE_TASKS, release all slots on the blacklisted task manager.

4. When a slot state changes from reserved(task assigned) to free(no task assigned), it will check whether the corresponding task manager is blacklisted. If yes, release the slot.

public interface BlacklistedTaskManagerChecker {
    /**
     * Returns whether the given task manager is blacklisted.
     */
    boolean isBlacklistedTaskManager(ResourceID resourceID);
}


public interface SlotPoolService {

    /** Start the encapsulated slot pool implementation. */
    void start(
            JobMasterId jobMasterId,
            String address,
            ComponentMainThreadExecutor mainThreadExecutor,
            BlacklistedTaskManagerChecker blacklistedTaskManagerChecker)
            throws Exception;

   /**
     * Releases all slots belonging to the owning TaskExecutor if it has been registered.
     *
     * @param taskManagerId identifying the TaskExecutor
     * @param cause cause for failing the slots
     */
    void releaseSlotsOnTaskManager(ResourceID taskManagerId, Exception cause);

    /**
     * Releases all free slots belonging to the owning TaskExecutor if it has been registered.
     *
     * @param taskManagerId identifying the TaskExecutor
     * @param cause cause for failing the slots
     */
    void releaseFreeSlotsOnTaskManager(ResourceID taskManagerId, Exception cause);

    //...
}

Blacklist on ResourceManager

ResourceManagerBlacklistHandler

ResourceManagerBlacklistHandler is a new component introduced in RM for the blacklist mechanism. It has only one sub-component: BlacklistTracker, which is responsible for managing cluster-level blacklisted items.

public interface ResourceManagerBlacklistHandler extends BlacklistHandler {
}

SlotManager

SlotManager should filter out blacklisted resources when allocating registered resources. To do that, we need following changes:

1. When starting SlotManager, the BlacklistedTaskManagerChecker will be passed in to check whether a registered task manager is blacklisted.
2. When trying to fulfill the slot requirements by registered task managers, the blacklisted ones will be filtered out.
3. SlotManager will request new task managers from external resource managers if the registered resources cannot fulfill the requirements. The blacklist also takes effect when requesting new task managers, the details will be described in ResourceManagerDriver.

public interface SlotManager {

    /** Starts the slot manager with the given leader id and resource manager actions. */
    void start(
            ResourceManagerId newResourceManagerId,
            Executor newMainThreadExecutor,
            ResourceActions newResourceActions,
            BlacklistedTaskManagerChecker newBlacklistedTaskManagerChecker);

    //...
}

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 blacklisted nodes, we need to pass the information of blacklisted nodes to ResourceManagerDriver at its initialization. 

ResourceManagerDriver uses following APIs to tell external resource managers about the information of blacklist nodes:

1. Yarn: AMRMClient#updateBlacklist 
2. Kubernetes: NodeAffinity

public interface BlacklistedNodeRetriever {

    /** Retrieve blacklisted nodes. */
    Set<String> getBlacklistedNodes();
}

public interface ResourceManagerDriver {

    /** Initialize the deployment specific components. */
    void initialize(
            ResourceEventHandler<WorkerType> resourceEventHandler,
            ScheduledExecutor mainThreadExecutor,
            Executor ioExecutor,
            BlacklistedNodeRetriever blacklistedNodeRetriever)
            throws Exception;

    //...
}

Synchronize Blacklist between JM & RM

The newly added or updated blacklisted items will be synchronized between JM and RM via RPC: 

1. Once a few blacklisted items are newly added (or updated) to the JobMasterBlacklistHandler, RM will be notified of these items via ResourceManagerGateway#notifyNewBlacklistedItems.
2. When RM receives the blacklisted items notified by a JM, it will add them into ResourceManagerBlacklistHandler, and notify all JMs of the successfully added (or updated) items through JobMasterGateway#notifyNewBlacklistedItems. 
3. Similarly, when JM receives the blacklisted items notified by RM, it will also add them to JobMasterBlacklistHandler.

public interface BlacklistListener {

    /** Notify new blacklisted items. */
    void notifyNewBlacklistedItems(Collection<BlacklistedItem<?>> newItems);
}

public interface JobManagerGateway extends BlacklistListener {
    //...
}

public interface ResourceManagerGateway extends BlacklistListener {     
    //...
}  

Enrich TaskManagerLocation with node information

In order to support blacklisting 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.

public class TaskManagerLocation {

    public String getNodeId();

    //...
}

Metrics

We propose to introduce following Gauge metrics to ResourceManagerMetricGroup:

1. numBlacklistedTaskMangers: The number of currently blacklisted task managers (including task managers on blacklisted nodes)
2. numBlacklistedNodes: The number of currently blacklisted nodes

REST API

We will introduce following REST APIs for blacklist mechanism:

1. REST API for querying blacklist information.
2. REST API for adding new blacklisted items.
3. REST API for removing existing blacklisted items.

Query

Add a REST API to obtain blacklist information. Each request will return all current blacklist items, which are obtained from ResourceManagerBlacklistHandler.

GET: http://{jm_rest_address:port}/blacklist

Request: {}

Response:

{
  /** This group only contains directly blacklisted task managers */
  "blacklistedTaskManagers": [
      {
          "id" : "container_XXX_000002",
          "timestamp" : "XXX",
          "action" : "MARK_BLACKLISTED"
      },
      {
          "id" : "container_XXX_000003",
          "timestamp" : "XXX",
          "action" : "MARK_BLACKLISTED"
      }, 
      ...
  ],
  "blacklistedNodes": [
      {
          "id" : "node1",
          "timestamp" : "XXX",
          "action" : "MARK_BLACKLISTED"
          "taskManagers" : [“container_XXX_000004”, “container_XXX_000005”, …]
      },
      ...
  ]
}

In order to get blacklist information, we need to add an interface to ResourceManagerGateway:

public interface ResourceManagerGateway {
   CompletableFuture<BlacklistInfo> requestBlacklist(@RpcTimeout Time timeout);
   // ...
}

Add

POST: http://{jm_rest_address:port}/blacklist/add

Request:

{
  "newBlacklistedTaskManagers": [
      {
          "id" : "container_XXX_000002",
          "action" : "MARK_BLACKLISTED"
      },
      {
          "id" : "container_XXX_000003",
          "action" : "MARK_BLACKLISTED"
      }, 
      ...
  ],
  "newBlacklistedNodes": [
      {
          "id" : "node1",
          "action" : "MARK_BLACKLISTED"
      },
      ...
  ]
}

Response: {}

Remove

POST: http://{jm_rest_address:port}/blacklist/remove

Request:

{
  "blacklistedTaskManagersToRemove": [
      {
          "id" : "container_XXX_000002",
          "action" : "MARK_BLACKLISTED"
      },
      {
          "id" : "container_XXX_000003",
          "action" : "MARK_BLACKLISTED"
      }, 
      ...
  ],
  "blacklistedNodesToRemove": [
      {
          "id" : "node1",
          "action" : "MARK_BLACKLISTED"
      },
      ...
  ]
}

Response: {}

Compatibility, Deprecation, and Migration Plan

The blacklist mechanism will be an optional feature which the user has to activate explicitly by setting the config option cluster.resource-blacklist.enabled: true. This entails that Flink's default behavior won't change.

Test Plan

1. The changes will be covered by unit and IT cases.
2. Test the functionality in a real Standanlone/Yarn/Kubernetes cluster.

Rejected Alternatives

No rejected alternatives yet.