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

/** A simple wrapper around UUID that abstracts a Process Id */
public class ProcessId {

    public ProcessId(final UUID id) {
        this.id = id;
    }

    public id() {
        return id;
    }
}
 

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

/**
 * A read-only metadata class representing the current state of each KafkaStreams client with at least one StreamThread participating in this rebalance
 */
public interface KafkaStreamsState {
  /**SortedSet<String> consumerClientIds();

  /**
   * @return the set of all active tasks owned by consumers on this KafkaStreams client since the previous rebalance
   */
  SortedSet<TaskId> previousActiveTasks();

  /**
   * @return the set of all standby tasks owned by consumers on this KafkaStreams client since the previous rebalance
   */
  SortedSet<TaskId> previousStandbyTasks();

  /**
   * Returns the total lag across all logged stores in the task. Equal to the end offset sum if this client
   * did not have any state for this task on disk.
   *
   * @return end offset sum - offset sum
   *          Task.LATEST_OFFSET if this was previously an active running task on this client
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
    */
  long lagFor(final TaskId task);

  /**
   * @return the previous tasks assigned to this consumer ordered by lag, filtered for any tasks that don't exist in this assignment
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
   */
  SortedSet<TaskId> prevTasksByLag(final String consumerClientId);

  /**
   * Returns a collection containing all (and only) stateful tasks in the topology by {@link TaskId},
   * mapped to its "offset lag sum". This is computed as the difference between the changelog end offset
   * and the current offset, summed across all logged state stores in the task.
   *
   * @return a map from all stateful tasks to their lag sum
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
   */
  Map<TaskId, Long> statefulTasksToLagSums();

  /**
   * The {@link HostInfo} of this KafkaStreams client, if set via the
   * {@link org.apache.kafka.streams.StreamsConfig#APPLICATION_SERVER_CONFIG application.server} config
   *
   * @return the processIdhost info offor thethis applicationKafkaStreams instanceclient runningif onconfigured, thiselse KafkaStreams client{@code Optional.empty()}
    */
  ProcessIDOptional<HostInfo> processIdhostInfo();

  /**
   * ReturnsThe theclient number of processing threads available to work on tasks tags for this KafkaStreams client, 
if set any *have whichbeen represents its overall capacity for work relative to other KafkaStreams clients.via configs using the
   * {@link org.apache.kafka.streams.StreamsConfig#clientTagPrefix}
   * <p>
   * @returnCan thebe numberused ofhowever processingyou threadswant, onor thispassed KafkaStreamsin client
to enable  */
  int numProcessingThreads();

  /*the rack-aware standby task assignor.
   *
   * @return all the setclient oftags consumerfound clientin ids for this KafkaStreams client's {@link org.apache.kafka.streams.StreamsConfig}
   */
  Map<String, SortedSet<String>String> consumerClientIdsclientTags();

  /**
   * @return the setrackId offor allthis activeKafkaStreams tasksclient, ownedor by consumers on this KafkaStreams client since the previous rebalance{@link Optional#empty()} if none was configured
   */
  SortedSet<TaskId>Optional<String> previousActiveTasksrackId();

  /**
   * @return the set of all standby tasks owned by consumers on this KafkaStreams client since the previous rebalance
   */
  SortedSet<TaskId> previousStandbyTasks();

  /**
   * Returns the total lag across all logged stores in the task. Equal to the end offset sum if this client
   * did not have any state for this task on disk.
   *
   * @return end offset sum - offset sum
   *          Task.LATEST_OFFSET if this was previously an active running task on this client
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
    */
  long lagFor(final TaskId task);

  /**
   * @return the previous tasks assigned to this consumer ordered by lag, filtered for any tasks that don't exist in this assignment
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
   */
  SortedSet<TaskId> prevTasksByLag(final String consumerClientId);

  /**
   * Returns a collection containing all (and only) stateful tasks in the topology by {@link TaskId},
   * mapped to its "offset lag sum". This is computed as the difference between the changelog end offset
   * and the current offset, summed across all logged state stores in the task.
   *
   * @return a map from all stateful tasks to their lag sum
   * @throws UnsupportedOperationException if the user did not request task lags be computed.
   */
  Map<TaskId, Long> statefulTasksToLagSums();

  /**
   * The {@link HostInfo} of this KafkaStreams client, if set via the
   * {@link org.apache.kafka.streams.StreamsConfig#APPLICATION_SERVER_CONFIG application.server} config
   *
   * @return the host info for this KafkaStreams client if configured, else {@code Optional.empty()}
   */
  Optional<HostInfo> hostInfo();

  /**
   * The client tags for this KafkaStreams client, if set any have been via configs using the
   * {@link org.apache.kafka.streams.StreamsConfig#clientTagPrefix}
   * <p>
   * Can be used however you want, or passed in to enable the rack-aware standby task assignor.
   *
   * @return all the client tags found in this KafkaStreams client's {@link org.apache.kafka.streams.StreamsConfig}
   */
  Map<String, String> clientTags();

  /**
   * @return the rackId for this KafkaStreams client, or {@link Optional#empty()} if none was configured
   */
  Optional<String> rackId();

  }

package org.apache.kafka.streams.processor.assignment;
 
/**
 * This is a simple container class used during the assignment process to distinguish
 * TopicPartitions type. Since the assignment logic can depend on the type of topic we're
 * looking at, and the rack information of the partition, this container class should have
 * everything necessary to make informed task assignment decisions.
 */
public interface TaskTopicPartition {
    /**
     *
     * @return the {@code TopicPartition} for this task.
     */
    TopicPartition topicPartition( * 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 KafkaStreams client assignments to add standby tasks to
     */
    public static void defaultStandbyTaskAssignment(final ApplicationState applicationState, 
                                                    final Map<ProcessId, KafkaStreamsAssignment> KafkaStreamsAssignments);

    /**
     *
 Optimize active task assignment *for @returnrack whetherawareness. theThis underlying topicoptimization is abased sourceon topic or not. Source changelog topicsthe 
     * {@link        are both source topics and changelog topics.StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} 
     * and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
     */
 configs which balance boolean isSource();

    /**cross rack traffic minimization and task movement.
     *
 Setting {@code trafficCost} to a *larger @returnnumber whetherreduces the underlyingoverall topiccross israck atraffic changelogof topicthe or not. Source changelog topicsresulting 
     * assignment, but can increase the number   are both source topics and changelog topics.of tasks shuffled around between clients. 
     */
    boolean isChangelog();

    /**
     * Setting {@code nonOverlapCost} to a larger number increases the affinity of tasks to their intended client
     * @returnand reduces the brokeramount rackby ids on which thisthe topicrack-aware partitionoptimization resides.can Ifshuffle notasks informationaround, could
at the cost of higher
 *    * cross-rack traffic.
   be found, this* willIn returnan anextreme emptycase, optionalif value.
we set {@code nonOverlapCost} to */
0 and @{code trafficCost} Optional<Set<String>> rackIds();
}

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.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>
     * If rack-aware client tags are configured, the rack-awareThis method optimizes cross-rack traffic for active tasks only. For standby task assignor will be usedoptimization,
     * use {@link #optimizeRackAwareStandbyTasks}.
     * <p>
     * @paramIt applicationStateis recommended to run this optimization before assigning theany metadatastandby andtasks, otherespecially infoif describingyou thehave currentconfigured
 application state
     * @paramyour KafkaStreamsAssignmentsKafkaStreams theclients currentwith assignment tags ofvia tasks to KafkaStreams clients
     *the rack.aware.assignment.tags config since this method may
     * shuffle @returnaround aactive newtasks mapwithout containingconsidering the mappingsclient fromtags KafkaStreamsAssignmentsand updatedcan withresult thein defaulta standbyviolation assignment
of     */the original
    public static* Map<ProcessID,client KafkaStreamsAssignment> defaultStandbyTaskAssignment(final ApplicationState applicationState, 
  tag assignment's constraints.
     *
     * @param kafkaStreamsAssignments the assignment of tasks to KafkaStreams clients to be optimized
     * @param optimizationParams                    optional configuration parameters to apply 
     */
    public static void optimizeRackAwareActiveTasks(final Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments,
                                                    final RackAwareOptimizationParams final Map<ProcessID, KafkaStreamsAssignment> KafkaStreamsAssignments);optimizationParams);      

    /**
     * Optimize activestandby 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>
     * This method optimizes cross-rack traffic for activestandby tasks only. For standbyactive task optimization,
     * use {@link #optimizeRackAwareStandbyTasks#optimizeRackAwareActiveTasks}.
     * 
     * @param applicationState        KafkaStreamsAssignments the metadatacurrent andassignment otherof infotasks describingto theKafkaStreams currentclients
 application state
   * @param  optimizationParams * @param kafkaStreamsAssignments the current assignmentoptional ofconfiguration tasksparameters to KafkaStreamsapply clients
     */
 @param tasks  public static void optimizeRackAwareStandbyTasks(final Map<ProcessId, KafkaStreamsAssignment> kafkaStreamsAssignments,
                     the  set  of  tasks  to  reassign  if  possible.  Must  already  be  assigned  to  a  KafkaStreams  client
  final    *RackAwareOptimizationParams optimizationParams);

     /**
  @return a new map* containingReturn the mappings from KafkaStreamsAssignments updated with the default rack-aware assignment for active tasks
     */
    public static Map<ProcessID, KafkaStreamsAssignment> optimizeRackAwareActiveTasks(final ApplicationState applicationState, 
         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 finalassignment Map<ProcessID,passes KafkaStreamsAssignment>the
 kafkaStreamsAssignments, 
   * 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 finaland SortedSet<TaskId>kill tasks);      

    /**the thread, so it may be useful to
     * Optimizeutilize standbythis taskmethod assignmentin foran rackassignor 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}to verify the assignment before returning it and fix any errors it finds.
     *
     * configs which balance cross rack traffic minimization and task movement@param applicationState The application for which this task assignment is being assessed.
     * Setting@param {@codetaskAssignment trafficCost} to aThe largertask numberassignment reducesthat thewill overallbe crossvalidated.
 rack traffic of the resulting *
     * @return assignment, but can increase the number of tasks shuffled around between clients. 
{@code AssignmentError.NONE} if the assignment created for this application is valid,
     *         *or Settinganother {@code nonOverlapCostAssignmentError} tootherwise.
 a larger number increases the*/
 affinity of tasks topublic theirstatic intended client
  AssignmentError validateTaskAssignment(final ApplicationState applicationState,
   * 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 willfinal haveTaskAssignment antaskAssignment) 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{
  }