Page History

package org.apache.kafka.streams.processor.assignment;

/**
 * A set of utilities to help implement task assignment
 */
public final class TaskAssignmentUtils {
    /**
     * Assign standby tasks to KafkaStreams clients according to the default logic.
     * <p>
     * If rack-aware client tags are configured, the rack-aware standby task assignor will be used
     *
     * @param applicationState        the metadata and other info describing the current application state
     * @param KafkaStreamsAssignments the currentKafkaStreams assignmentclient ofassignments tasks to KafkaStreamsadd clients
standby tasks    *
     * @return a new map containing the mappings from KafkaStreamsAssignments updated with the default standby assignment
 to
     */
    public static Map<ProcessID,void KafkaStreamsAssignment> defaultStandbyTaskAssignment(final ApplicationState applicationState, 
                                                                                      final Map<ProcessIDMap<ProcessId, KafkaStreamsAssignment> KafkaStreamsAssignments);

    /**
     * Optimize active task assignment for rack awareness. This optimization is based on the 
     * {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} 
     * and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
     * configs which balance cross rack traffic minimization and task movement.
     * Setting {@code 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 {@code 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 {@code 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 {@code trafficCost} to 0,
     * and {@code nonOverlapCost} to a positive value, the resulting assignment will be identical to the input assignment.    
     * <p>
     * Note: this method will modify the input {@link KafkaStreamsAssignment} objects and return the same map.
     * It does not make a copy of the map or the KafkaStreamsAssignment objects.      
     * <p>
     * This method optimizes cross-rack traffic for active tasks only. For standby task optimization,
     * use {@link #optimizeRackAwareStandbyTasks}.
     * <p>
     * @paramIt applicationStateis recommended to run this optimization before assigning theany metadatastandby andtasks, otherespecially infoif describingyou the current application statehave configured
     * @paramyour kafkaStreamsAssignmentsKafkaStreams theclients currentwith assignment oftags tasksvia to KafkaStreams clients
     * @param tasksthe rack.aware.assignment.tags config since this method may
     * shuffle around active tasks without considering the client tags and can result in a theviolation setof ofthe tasksoriginal
 to reassign if possible. Must* alreadyclient betag assigned to a KafkaStreams clientassignment's constraints.
     *
     * @param kafkaStreamsAssignments the assignment  
     * @return a map with the KafkaStreamsAssignments updated to minimize cross-rack traffic for active tasksof tasks to KafkaStreams clients to be optimized
     * @param optimizationParams      optional configuration parameters to apply 
     */
    public static Map<ProcessID, KafkaStreamsAssignment>void optimizeRackAwareActiveTasks(final ApplicationState applicationState, Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments,
                                                    final RackAwareOptimizationParams optimizationParams);      

    /**
     * Optimize standby task assignment for rack awareness. This optimization is based on the 
     * {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} 
     * and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
     * configs which balance cross rack traffic minimization and final Map<ProcessID, KafkaStreamsAssignment> kafkaStreamsAssignments, 
                   task movement.
     * Setting {@code 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 {@code 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
     final* SortedSet<TaskId> tasks);      

    /**
     * Optimize standby task assignment for rack awareness. This optimization is based on the 
     * {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} cross-rack traffic.
     * In an extreme case, if we set {@code 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 {@code trafficCost} to 0,
     * and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG@code nonOverlapCost}
 to a positive value, *the configsresulting whichassignment balancewill crossbe rackidentical trafficto minimizationthe andinput task movementassignment.
     * Setting<p>
 {@code trafficCost} to a larger* numberThis reducesmethod the overalloptimizes cross -rack traffic offor thestandby resultingtasks 
only. For active task  * assignmentoptimization,
 but can increase the number* ofuse tasks shuffled around between clients. {@link #optimizeRackAwareActiveTasks}.
     * Setting
 {@code nonOverlapCost} to a larger* number@param increasesKafkaStreamsAssignments the current affinityassignment of tasks to theirKafkaStreams intended clientclients
     * @param *optimizationParams and reduces the amount by whichoptional theconfiguration rack-awareparameters optimizationto canapply shuffle
 tasks around, at the cost of higher    */
    public static void optimizeRackAwareStandbyTasks(final Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments,
                                                     final RackAwareOptimizationParams optimizationParams);

    /**
     * Return a "no-op" assignment that just copies the previous assignment of tasks to KafkaStreams clients
     *
     * cross-rack traffic. @param applicationState the metadata and other info describing the current application state
     *
 In an  extreme * case,@return ifa wenew setmap {@codecontaining nonOverlapCost}an toassignment 0that andreplicates @{code trafficCost} to a positive value,
     * the resulting assignment will have an absolute minimum of cross rack traffic. If we set {@code trafficCost} to 0,
     * and {@code nonOverlapCost} to a positive value, the resulting assignment will be identical to the input assignment.
     * <p>exactly the previous assignment reported in the applicationState
     */
    public static Map<ProcessId, KafkaStreamsAssignment> identityAssignment(final ApplicationState applicationState);

    /**
     * Validate the passed-in {@link TaskAssignment} and return an {@link AssignmentError} representing the
     * first error detected in the assignment, or {@link AssignorError.NONE} if the assignment passes the
     * Thisverification methodcheck.
 optimizes cross-rack traffic for standby* tasks<p>
 only. For active task optimization,
* Note: this verification is *performed useautomatically {@link #optimizeRackAwareActiveTasks}.
     * by the StreamsPartitionAssignor on the assignment
     * @paramreturned KafkaStreamsAssignmentsby the TaskAssignor, currentand assignmentthe oferror tasksreturned to KafkaStreamsthe clients
assignor via the {@link TaskAssignor#onAssignmentComputed}
 * @param applicationState  * callback. Therefore, it is not therequired metadatato andcall otherthis infomanually describingfrom the current{@link applicationTaskAssignor#assign} statemethod.
     *
 However  if  *an @returninvalid aassignment newis mapreturned containingit thewill mappingsfail fromthe KafkaStreamsAssignmentsrebalance updatedand withkill the default rack-aware assignment for standy tasks
     */
    public static Map<ProcessID, KafkaStreamsAssignment> optimizeRackAwareStandbyTasks(final ApplicationState applicationState,
                     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.
     *
     * @param applicationState The application for which this task assignment is being assessed.
     * @param taskAssignment   The task assignment that will be validated.
     *
     * @return {@code AssignmentError.NONE} if the assignment created for this application is valid,
     * final Map<ProcessID, KafkaStreamsAssignment> kafkaStreamsAssignments);

    /**
   or another * Return a "no-op" assignment that just copies the previous assignment of tasks to KafkaStreams clients{@code AssignmentError} otherwise.
     */
    public static AssignmentError validateTaskAssignment(final ApplicationState applicationState,
     *
     * @param applicationState the metadata and other info describing the current application state
     *
     * @return a new map containing an assignment that replicates exactly the previous assignment reported in the applicationState
     */
    public static Map<ProcessID, KafkaStreamsAssignment> identityAssignment(final ApplicationStateTaskAssignment applicationStatetaskAssignment); {
  }