Class TaskAssignmentUtils

java.lang.Object
org.apache.kafka.streams.processor.assignment.TaskAssignmentUtils
public final class TaskAssignmentUtils extends Object
A set of utilities to help implement task assignment via the TaskAssignor

  • Method Details

    • validateTaskAssignment

      public static TaskAssignor.AssignmentError validateTaskAssignment(ApplicationState applicationState, TaskAssignor.TaskAssignment taskAssignment)
      Validate the passed-in TaskAssignor.TaskAssignment and return an TaskAssignor.AssignmentError representing the first error detected in the assignment, or TaskAssignor.AssignmentError.NONE if the assignment passes the verification check.

      Note: this verification is performed automatically by the StreamsPartitionAssignor on the assignment returned by the TaskAssignor, and the error returned to the assignor via the TaskAssignor.onAssignmentComputed(ConsumerPartitionAssignor.GroupAssignment, ConsumerPartitionAssignor.GroupSubscription, TaskAssignor.AssignmentError) callback. Therefore, it is not required to call this manually from the TaskAssignor.assign(ApplicationState) method. However, if an invalid assignment is returned it will fail the rebalance and kill the thread, so it may be useful to utilize this method in an assignor to verify the assignment before returning it and fix any errors it finds.

      Parameters:
      applicationState - The application for which this task assignment is being assessed.
      taskAssignment - The task assignment that will be validated.
      Returns:
      AssignmentError.NONE if the assignment created for this application is valid, or another AssignmentError otherwise.

    • identityAssignment

      public static Map<ProcessId, KafkaStreamsAssignment> identityAssignment(ApplicationState applicationState)
      Return a "no-op" assignment that just copies the previous assignment of tasks to KafkaStreams clients
      Parameters:
      applicationState - the metadata and other info describing the current application state
      Returns:
      a new map containing an assignment that replicates exactly the previous assignment reported in the applicationState

    • defaultStandbyTaskAssignment

      public static void defaultStandbyTaskAssignment(ApplicationState applicationState, Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments)
      Assign standby tasks to KafkaStreams clients according to the default logic.

      If rack-aware client tags are configured, the rack-aware standby task assignor will be used

      Parameters:
      applicationState - the metadata and other info describing the current application state
      kafkaStreamsAssignments - the KafkaStreams client assignments to add standby tasks to

    • optimizeRackAwareActiveTasks

      public static void optimizeRackAwareActiveTasks(TaskAssignmentUtils.RackAwareOptimizationParams optimizationParams, Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments)
      Optimize active task assignment for rack awareness. This optimization is based on the trafficCost and nonOverlapCost configs which balance cross rack traffic minimization and task movement. Setting trafficCost to a larger number reduces the overall cross rack traffic of the resulting assignment, but can increase the number of tasks shuffled around between clients. Setting nonOverlapCost to a larger number increases the affinity of tasks to their intended client and reduces the amount by which the rack-aware optimization can shuffle tasks around, at the cost of higher cross-rack traffic. In an extreme case, if we set nonOverlapCost to 0 and @{code trafficCost} to a positive value, the resulting assignment will have an absolute minimum of cross rack traffic. If we set trafficCost to 0, and nonOverlapCost to a positive value, the resulting assignment will be identical to the input assignment.

      Note: this method will modify the input KafkaStreamsAssignment objects and return the same map. It does not make a copy of the map or the KafkaStreamsAssignment objects.

      This method optimizes cross-rack traffic for active tasks only. For standby task optimization, use optimizeRackAwareStandbyTasks(TaskAssignmentUtils.RackAwareOptimizationParams, Map).

      It is recommended to run this optimization before assigning any standby tasks, especially if you have configured your KafkaStreams clients with assignment tags via the rack.aware.assignment.tags config since this method may shuffle around active tasks without considering the client tags and can result in a violation of the original client tag assignment's constraints.

      Parameters:
      optimizationParams - optional configuration parameters to apply
      kafkaStreamsAssignments - the current assignment of tasks to KafkaStreams clients

    • optimizeRackAwareStandbyTasks

      public static void optimizeRackAwareStandbyTasks(TaskAssignmentUtils.RackAwareOptimizationParams optimizationParams, Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments)
      Optimize standby task assignment for rack awareness. This optimization is based on the trafficCost and nonOverlapCost configs which balance cross rack traffic minimization and task movement. Setting trafficCost to a larger number reduces the overall cross rack traffic of the resulting assignment, but can increase the number of tasks shuffled around between clients. Setting nonOverlapCost to a larger number increases the affinity of tasks to their intended client and reduces the amount by which the rack-aware optimization can shuffle tasks around, at the cost of higher cross-rack traffic. In an extreme case, if we set nonOverlapCost to 0 and @{code trafficCost} to a positive value, the resulting assignment will have an absolute minimum of cross rack traffic. If we set trafficCost to 0, and nonOverlapCost to a positive value, the resulting assignment will be identical to the input assignment.

      Note: this method will modify the input KafkaStreamsAssignment objects and return the same map. It does not make a copy of the map or the KafkaStreamsAssignment objects.

      This method optimizes cross-rack traffic for standby tasks only. For active task optimization, use optimizeRackAwareActiveTasks(TaskAssignmentUtils.RackAwareOptimizationParams, Map).

      Parameters:
      optimizationParams - optional configuration parameters to apply
      kafkaStreamsAssignments - the current assignment of tasks to KafkaStreams clients