Page History

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

public interface TaskAssignor extends Configurable {

  /**
   * @param applicationMetadata the metadata for this Streams application
   * @return the assignment of active and standby tasks to Streams client nodes
   */
  TaskAssignment assign(ApplicationMetadata applicationMetadata);

  /**
   * This method allows the assignor plugin to observe the final assignment returned
   * to the brokers.
   * @param assignment: the final assignment returned to the kafka broker
   * @param subscription: the original subscription passed into the assignor
   */
  void finalAssignmentonAssignmentComputed(GroupAssignment assignment, GroupSubscription subscription);

  /**
   * Wrapper class for the final assignment of active and standbys tasks to individual Streams 
   * client nodes
   */
  class TaskAssignment {
    private final Collection<NodeAssignment>Collection<StreamsClientAssignment> nodeAssignmentsstreamsClientAssignments;

	/**
     * @return the assignment of tasks to kafka nodesstreams clients
     */
    public Collection<NodeAssignment>Collection<StreamsClientAssignment> assignment();

    /**
     * @return the number of Streams client nodes to which tasks were assigned
     */
    public int numNodesnumStreamsClients();

    /**
     * @return a String representation of the returned assignment, in processId order
     */
    @Override
    public String toString();
  }
}

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

/**
 * A simple interface for the assignor to return the desired placement of active and standby tasks on Streams client nodes 
 */
public interface NodeAssignmentStreamsClientAssignment {
  ProcessID processId();

  long followupRebalanceDeadline();

  Map<TaskId, AssignedTask> assignment();

  /**
   * @return the actual deadline in objective time, using ms since the epoch, after which the
   * followup rebalance will be attempted. Equivalent to {@code 'now + followupRebalanceDelay'}
   */
  long followupRebalanceDeadline();

  static class AssignedTask {
    public final boolean isActive;
    public final boolean isStateful;
  }
}

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

/**
 * A read-only metadata class representing the current state of each Streams client node with at least one StreamThread participating in this rebalance
 */
public interface ApplicationMetadata {
    /**
     * @return a map from the {@code processId} to {@link NodeStateStreamsClientState} for all Streams client nodes in this app
     */
    Map<ProcessID, NodeState>StreamsClient> nodeStatesstreamsClientStates();

    /**
     * Makes a remote call to fetch changelog topic end offsets and, if successful, uses the results to compute
     * task lags for each {@link NodeStateStreamsClientState}.
     *
     * @return whether the end offset fetch and lag computation was successful
     */
    boolean computeTaskLags();

    /**
     * @return a simple container class with the Streams configs relevant to assignment
     */
    AssignmentConfigs assignmentConfigs();

    /**
     * @return the set of all tasks in this topology which must be assigned to a node
     */
    Set<TaskId> allTasks();

    /**
     *
     * @return the set of stateful and changelogged tasks in this topology
     */
    Set<TaskId> statefulTasks(); 
}

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

/**
 * A set of utilities to help implement task assignment
 */
public final class TaskAssignmentUtils {
    /**
     * Assign standby tasks to nodes according to the default logic.
     * <p>
     * If rack-aware client tags are configured, the rack-aware standby task assignor will be used
     *
     * @param nodeAssignmentsstreamsClientAssignments the current assignment of tasks to nodes
     */
    public static void defaultStandbyTaskAssignment(final ApplicationMetadata applicationMetadata, final Map<ProcessID, NodeAssignment>StreamsClientAssignment> nodeAssignmentsstreamsClientAssignments) {...}

    /**
     * Optimize the active task assignment for rack-awareness
     *
     * @param nodeAssignments the current assignment of tasks to nodes
     * @param tasks the set of tasks to reassign if possible. Must already be assigned to a node
     */
    public static void optimizeRackAwareActiveTasks(final ApplicationMetadata applicationMetadata, final Map<ProcessID, NodeAssignment>StreamsClientAssignment> nodeAssignmentsstreamsClientAssignments, final SortedSet<TaskId> tasks) {...}

    /**
     * Optimize the standby task assignment for rack-awareness
     *
     * @param nodeAssignmentsstreamsClientAssignments the current assignment of tasks to nodes
     */
    public static void optimizeRackAwareStandbyTasks(final ApplicationMetadata applicationMetadata, final Map<ProcessID, NodeAssignment>StreamsClientAssignment> nodeAssignmentsstreamsClientAssignments) {...}
}