content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.streams.processor.assignment;
  
  import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.GroupAssignment;
  import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.GroupSubscription;
  import org.apache.kafka.common.Configurable;
  import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
  import org.apache.kafka.streams.errors.TaskAssignmentException;
  
  import java.util.Collection;
  import java.util.Map;
  
  /**
   * A TaskAssignor is responsible for creating a TaskAssignment from a given
   * {@code ApplicationState}.
   * The implementation may also override the {@code onAssignmentComputed} callback for insight into
   * the result of the assignment result.
   */
  public interface TaskAssignor extends Configurable {
  
      /**
       * NONE: no error detected
       * ACTIVE_TASK_ASSIGNED_MULTIPLE_TIMES: multiple KafkaStreams clients assigned with the same active task
       * INVALID_STANDBY_TASK: stateless task assigned as a standby task
       * MISSING_PROCESS_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,
          INVALID_STANDBY_TASK,
          MISSING_PROCESS_ID,
          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>
       * Note: some kinds of errors will make it impossible for the StreamsPartitionAssignor to parse the TaskAssignment
       * that was returned from the 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 kafka broker, or an empty assignment map if
       *                     an error prevented the assignor from converting the TaskAssignment into a GroupAssignment
       * @param subscription 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) {}
  
      @Override
      default void configure(final Map<String, ?> configs) {}
  
      /**
       * Wrapper class for the final assignment of active and standbys tasks to individual
       * KafkaStreams clients.
       */
      class TaskAssignment {
          private final Collection<KafkaStreamsAssignment> assignment;
  
          public TaskAssignment(final Collection<KafkaStreamsAssignment> assignment) {
              this.assignment = assignment;
          }
  
          /**
           * @return the assignment of tasks to kafka streams clients.
           */
          public Collection<KafkaStreamsAssignment> assignment() {
              return assignment;
          }
      }
  }