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

public interface TaskAssignor extends Configurable {    

   /**
     * NONE: no error detected
     * ACTIVE_TASK_ASSIGNED_MULTIPLE_TIMES: multiple KafkaStreams clients assigned with the same active task
     * ACTIVEINVALID_AND_STANDBY_TASK_ASSIGNED_TO_SAME_KAFKASTREAMS: activestateless task and standby task assigned toas thea samestandby KafkaStreams clienttask
     * INVALIDMISSING_STANDBYPROCESS_TASK: stateless task assigned as a standby task
     ID: ProcessId present in the input ApplicationState was not present in the output TaskAssignment
     * UNKNOWN_PROCESS_ID: unrecognized ProcessId not matching any of the participating consumers
     * UNKNOWN_TASK_ID: unrecognized TaskId not matching any of the tasks to be assigned
     */   
    enum AssignmentError {     
	    NONE,
	    ACTIVE_TASK_ASSIGNED_MULTIPLE_TIMES,     
	    ACTIVE_ANDINVALID_STANDBY_TASK_ASSIGNED_TO_SAME_KAFKASTREAMS,
	        INVALIDMISSING_STANDBYPROCESS_TASKID,
        UNKNOWN_PROCESS_ID,
	    UNKNOWN_TASK_ID
  }

  /**
   * @param applicationState the metadata for this Kafka Streams application
   *
   * @return the assignment of active and standby tasks to KafkaStreams clients 
   *
   * @throws TaskAssignmentException If an error occurs during assignment and you wish for the rebalance to be retried,
   *                                 you can throw this exception to keep the assignment unchanged and automatically
   *                                 schedule an immediate followup rebalance. 
   */
  TaskAssignment assign(ApplicationState applicationState);

  /**
   * This callback can be used to observe the final assignment returned to the brokers and check for any errors that 
   * were detected while processing the returned assignment. If any errors were found, the corresponding 
   * will be returned and a StreamsException will be thrown after this callback returns. The StreamsException will
   * be thrown up to kill the StreamThread and can be handled as any other uncaught exception would if the application
   * has registered a {@link StreamsUncaughtExceptionHandler}.
   * <p>
   * @param assignmentNote: some kinds the final assignment returned toof errors will make it impossible for the kafkaStreamsPartitionAssignor broker
to parse the *TaskAssignment
 @param subscription: the* originalthat subscriptionwas passedreturned intofrom the assignor
   * @param error:        the corresponding error type if one was detected while processing the returned assignment,  TaskAssignor's {@link #assign}. If this occurs, the {@link GroupAssignment} passed
   * in to this callback will contain an empty map instead of the consumer assignments.
   * 
   *  @param  assignment:      the  final consumer  assignments  returned  to  the  or AssignmentError.NONE if the returnedkafka broker, or an empty assignment wasmap validif
   */
  default void onAssignmentComputed(GroupAssignment assignment, GroupSubscription subscription, AssignmentError error) {}

  /**
   * Wrapper class for the final assignment ofan activeerror andprevented standbysthe tasksassignor tofrom individualconverting 
the TaskAssignment into *a KafkaStreams clientsGroupAssignment
   */
 @param classsubscription: TaskAssignment {

	/**
     * @return the assignment of tasks to kafka streams clients
     */
    public Collection<KafkaStreamsAssignment> assignment();
  }
}the original consumer subscriptions passed into the assignor
   * @param error:        the corresponding error type if one was detected while processing the returned assignment,  
   *                      or AssignmentError.NONE if the returned assignment was valid
   */
  default void onAssignmentComputed(GroupAssignment assignment, GroupSubscription subscription, AssignmentError error) {}

  /**
   * Wrapper class for the final assignment of active and standbys tasks to individual 
   * KafkaStreams clients
   */
  class TaskAssignment {

	/**
     * @return the assignment of tasks to kafka streams clients
     */
    public Collection<KafkaStreamsAssignment> assignment();
  }
}

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 to
     */
    public *static @return a new map containing the mappings from KafkaStreamsAssignments updated with the default standby assignment
     */
    public static Map<ProcessID, KafkaStreamsAssignment> defaultStandbyTaskAssignment(void defaultStandbyTaskAssignment(final ApplicationState applicationState, 
                                                                                      final Map<ProcessIDfinal Map<ProcessId, KafkaStreamsAssignment> KafkaStreamsAssignments);

    /**
     * Optimize the active task assignment for rack-awareness
     * awareness. This optimization is based on the 
     * @param applicationState  {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} 
     * the metadata and other info describing the current application state{@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
     * @paramconfigs kafkaStreamsAssignmentswhich thebalance currentcross assignmentrack oftraffic tasksminimization toand KafkaStreamstask clientsmovement.
       * @paramSetting tasks{@code trafficCost} to a larger number reduces the overall cross rack traffic of the resulting 
    the set* ofassignment, tasksbut tocan reassignincrease ifthe possible.number Mustof alreadytasks beshuffled assignedaround tobetween aclients. KafkaStreams client
     *
 Setting    * @return{@code nonOverlapCost} to a newlarger mapnumber containingincreases the mappingsaffinity fromof KafkaStreamsAssignmentstasks updatedto withtheir theintended defaultclient
 rack-aware assignment for active tasks
* and    */
    public static Map<ProcessID, KafkaStreamsAssignment> optimizeRackAwareActiveTasks(final ApplicationState applicationState, 
    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.    
    final Map<ProcessID, KafkaStreamsAssignment> kafkaStreamsAssignments, * <p>
     * This method optimizes cross-rack traffic for active tasks only. For standby task optimization,
     * use {@link #optimizeRackAwareStandbyTasks}.
     * <p>
     * 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 finalclient SortedSet<TaskId> tasks);

    /**
     * Optimize the standby task assignment for rack-awareness
     tags and can result in a violation of the original
     * client tag assignment's constraints.
     *
     * @param KafkaStreamsAssignmentskafkaStreamsAssignments the current assignment of tasks to KafkaStreams clients
 to be optimized
     * @param optimizationParams applicationState     optional configuration parameters to theapply metadata
 and  other  info*/
 describing the current applicationpublic state
static void optimizeRackAwareActiveTasks(final Map<ProcessId, KafkaStreamsAssignment> *kafkaStreamsAssignments,
      * @return  a  new  map  containing  the  mappings  from  KafkaStreamsAssignments  updated  with  the  default rack-aware assignment for standy tasks
                    */
  final RackAwareOptimizationParams optimizationParams); public  static Map<ProcessID, KafkaStreamsAssignment> optimizeRackAwareStandbyTasks(final ApplicationState applicationState,   

    /**
     * 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 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 finaltasks Map<ProcessID,shuffled KafkaStreamsAssignment> kafkaStreamsAssignments);

    /**around between clients. 
     * Return a "no-op" assignment that just copiesSetting {@code nonOverlapCost} to a larger number increases the previousaffinity assignment of tasks to their KafkaStreamsintended clientsclient
     *
 and reduces the amount *by @paramwhich applicationState the metadata and other info describingrack-aware optimization can shuffle tasks around, at the currentcost applicationof statehigher
     * cross-rack traffic.
       * In @returnan aextreme newcase, mapif containingwe anset assignment{@code thatnonOverlapCost} replicatesto exactly0 theand previous assignment reported in the applicationState
     */
    public static Map<ProcessID, KafkaStreamsAssignment> identityAssignment(final ApplicationState applicationState);
 }

@{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>
     * This method optimizes cross-rack traffic for standby tasks only. For active task optimization,
     * use {@link #optimizeRackAwareActiveTasks}.
     * 
     * @param KafkaStreamsAssignments the current assignment of tasks to KafkaStreams clients
     * @param optimizationParams      optional configuration parameters to apply 
     */
    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
     *
     * @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 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
     * verification check.
     * <p>
     * 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 {@link TaskAssignor#onAssignmentComputed}
     * callback. Therefore, it is not required to call this manually from the {@link TaskAssignor#assign} 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.
     *
     * @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,
     *         or another {@code AssignmentError} otherwise.
     */
    public static AssignmentError validateTaskAssignment(final ApplicationState applicationState,
                                                         final TaskAssignment taskAssignment) {
  }