Page History

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

/**
 * A simple interface for the assignor to return the desired placement of active and standby tasks on KafkaStreams clients
  */
public interface KafkaStreamsAssignment {
  ProcessID processId();

  Set<AssignedTask> assignment();

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

  static class AssignedTask {

    public AssignedTask(final TaskId id, final Type taskType);

    enum Type {
        STATELESS,
        STATEFUL,
        STANDBY
    }
    
    public Type type();

    public TaskId 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 {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 {
  /**
   * @return the processId of the application instance running on this KafkaStreams client
   */
  ProcessID processId();

  /**
   * Returns the number of processing threads available to work on tasks for this KafkaStreams client, 
   * which represents its overall capacity for work relative to other KafkaStreams clients.
   *
   * @return the number of processing threads on this KafkaStreams client
   */
  int numProcessingThreads();

  /**
   * @return the set of consumer client ids for this KafkaStreams client
   */
  SortedSet<String> consumerClientIds();

  /**
   * @return the processIdset of all active thetasks applicationowned instanceby runningconsumers on this KafkaStreams client since the previous rebalance
    */
  ProcessIDSortedSet<TaskId> processIdpreviousActiveTasks();

  /**
   * Returns@return the numberset of consumerall clientsstandby intasks theowned consumerby groupconsumers foron this KafkaStreams client, 
since the  * which represents its overall capacity.previous rebalance
   * <p>/
   * NOTE: this is actually the "minimum capacity" of a KafkaStreams client, or the minimum number of assigned
   * active tasks below which the KafkaStreams client will have been over-provisioned and unable to give every
   * available consumer an active task assignmentSortedSet<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 the number@return ofend consumersoffset onsum this- KafkaStreamsoffset clientsum
   */
     int numConsumerClients();

  /**
  Task.LATEST_OFFSET *if @returnthis thewas setpreviously ofan consumeractive clientrunning idstask foron this KafkaStreams client
   */
  SortedSet<String>long consumerClientIdslagFor(final TaskId task);

  /**
   * @return the set of all active tasks owned by consumers on this KafkaStreams client since the previous rebalance previous tasks assigned to this consumer ordered by lag, filtered for any tasks that don't exist in this assignment
   */
  SortedSet<TaskId> previousActiveTasksprevTasksByLag(final String consumerClientId);

  /**
   * @returnReturns thea setcollection ofcontaining all standby(and tasksonly) ownedstateful bytasks consumers on this KafkaStreams client since in the previoustopology rebalance
   */
  SortedSet<TaskId> previousStandbyTasks();

  /**by {@link TaskId},
   * Returnsmapped to theits total"offset lag acrosssum". allThis loggedis storescomputed inas the task.difference Equalbetween tothe thechangelog end offset
 sum if this* client
and the current *offset, didsummed notacross haveall anylogged state forstores in thisthe task on disk.
   *
   * @return end offset sum - offset a map from all stateful tasks to their lag sum
   */
  Map<TaskId, Long> statefulTasksToLagSums();

  /**
   * Task.LATEST_OFFSET if this was previously an active running task on this clientThe {@link HostInfo} of this KafkaStreams client, if set via the
   */
  long lagFor(final TaskId task);

  /*{@link org.apache.kafka.streams.StreamsConfig#APPLICATION_SERVER_CONFIG application.server} config
   *
   * @return the previoushost tasksinfo assignedfor to this consumerKafkaStreams orderedclient byif lagconfigured, filteredelse for any tasks that don't exist in this assignment{@code Optional.empty()}
   */
  SortedSet<TaskId>Optional<HostInfo> prevTasksByLaghostInfo(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 offsetThe client tags for this KafkaStreams client, if set any have been via configs using the
   * {@link org.apache.kafka.streams.StreamsConfig#clientTagPrefix}
   * <p>
   * and the current offset, summed across all logged state stores in the taskCan be used however you want, or passed in to enable the rack-aware standby task assignor.
   *
   * @return aall the mapclient fromtags allfound statefulin tasksthis toKafkaStreams their lag sumclient's {@link org.apache.kafka.streams.StreamsConfig}
   */
  Map<TaskId, Long> statefulTasksToLagSums();

  /**
   * The {@link HostInfo} of this KafkaStreams client, if set via the
   * {@linkMap<String, String> clientTags();
 }

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 ApplicationState {
    /** in "true" will result in a remote call to fetch changelog topic end offsets and you should pass in "false" unless
     * you specifically need the task lag information.
     *
     * @return a map from the {@code processId} to {@link KafkaStreamsState} for all KafkaStreams clients in this app
     *
     * @throws TaskAssignmentException if a retriable error occurs while computing KafkaStreamsState metadata. Re-throw
     * @param  computeTaskLags whether or not to include task lag information in the returned metadata. Note that passing 
     * in "true" will result in a remote call tothis fetchexception changelogto topichave endKafka offsetsStreams andretry youthe shouldrebalance passby inreturning "false"the unlesssame
     *      you specifically need the task lag information.
     *
     * @return a map from the {@code processId} to {@link KafkaStreamsState} forassignment alland KafkaStreamsscheduling clientsan inimmediate thisfollowup apprebalance
     */
    Map<ProcessID, KafkaStreamsState> kafkaStreamsStates(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
     */
    Set<TaskId> allTasks();

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

    /**
     *
     * @return the set of stateless or changelog-less tasks in this topology
     */
    Set<TaskId> statelessTasks();

}