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 {
  /**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 the number of processing threads available to work on tasks client tags for this KafkaStreams client, 
if set any *have which represents its overall capacity for work relative to other KafkaStreams clients.been 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 client idsin 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( * 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 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 underlyingoptimization topic is abased sourceon topic or not. Source changelog topicsthe 
     * {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} 
     are* bothand source topics and changelog topics.{@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
     */
 configs which balance cross boolean isSource();

    /**rack traffic minimization and task movement.
     *
 Setting {@code trafficCost} to a *larger @returnnumber whetherreduces the underlying topic is a changelog topic or not. Source changelog topics
     *         are both source topics and changelog topics.
     */
    boolean isChangelog();

    /**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 @returnwhich the broker rack-aware idsoptimization oncan whichshuffle thistasks topicaround, partitionat resides.the Ifcost noof information couldhigher
     * cross-rack traffic.
     * In an beextreme foundcase, this will return an empty optional value.
     */
if we set {@code nonOverlapCost} to 0 and @{code trafficCost} to a positive value,
     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.* 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-aware standby task assignor will be usedThis 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 thehave currentconfigured
 application state
     * @paramyour KafkaStreamsAssignmentsKafkaStreams theclients currentwith assignment oftags tasksvia to KafkaStreams clients
     *the rack.aware.assignment.tags config since this method may
     * @returnshuffle around aactive newtasks mapwithout containingconsidering the mappings from KafkaStreamsAssignments updated with the default standby assignment
     */
    public static Map<ProcessID, KafkaStreamsAssignment> defaultStandbyTaskAssignment(final ApplicationState applicationState, 
      client tags and can result in a violation of the original
     * client 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 Map<ProcessID, KafkaStreamsAssignment> KafkaStreamsAssignments);                           final RackAwareOptimizationParams 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 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.
     * the<p>
 resulting assignment will have an* absoluteThis minimummethod ofoptimizes cross -rack traffic. for Ifstandby wetasks setonly. {@codeFor trafficCost}active totask 0optimization,
     * anduse {@code@link nonOverlapCost} to a positive value,#optimizeRackAwareActiveTasks}.
     * 
     * @param KafkaStreamsAssignments the resultingcurrent assignment will be identical to the input assignment.
of tasks to KafkaStreams clients
     * @param optimizationParams     * <p>
optional configuration parameters to apply *
 This method optimizes cross-rack traffic for active tasks only. For standby task optimization,
     * use {@link #optimizeRackAwareStandbyTasks}.
     *
     * @param applicationState        the metadata and other info describing the current application state
     * @param kafkaStreamsAssignments the current   */
    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 tasks*
     * @param applicationState the metadata and other info      describing the setcurrent of tasks to reassign if possible. Must already be assigned to a KafkaStreams clientapplication state
     *
     * @return a new map containing thean mappingsassignment fromthat KafkaStreamsAssignmentsreplicates updatedexactly with the defaultprevious rack-awareassignment assignmentreported forin activethe tasksapplicationState
     */
    public static Map<ProcessIDMap<ProcessId, KafkaStreamsAssignment> optimizeRackAwareActiveTasksidentityAssignment(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
      final Map<ProcessID, KafkaStreamsAssignment> kafkaStreamsAssignments, 
                * 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 finalfinds.
 SortedSet<TaskId> tasks);      *

     /**
 @param applicationState The application *for Optimizewhich standbythis task assignment foris rackbeing awarenessassessed.
 This optimization is based on* the@param 
taskAssignment   The task *assignment {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_TRAFFIC_COST_CONFIG trafficCost} that will be validated.
     *
 and {@link StreamsConfig#RACK_AWARE_ASSIGNMENT_NON_OVERLAP_COST_CONFIG nonOverlapCost}
 * @return {@code AssignmentError.NONE} *if configsthe whichassignment balancecreated crossfor rackthis trafficapplication minimization and task movement.is valid,
     * Setting {@code trafficCost} to a larger number reduces theor overallanother cross{@code rack traffic of the resulting AssignmentError} otherwise.
     */
 assignment, but can increasepublic thestatic numberAssignmentError ofvalidateTaskAssignment(final tasksApplicationState shuffledapplicationState,
 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.
   final TaskAssignment *taskAssignment) 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 standby tasks only. For active task optimization,{
  }