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.internals;
  
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.OffsetAndMetadata;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.streams.errors.LockException;
  import org.apache.kafka.streams.errors.StreamsException;
  import org.apache.kafka.streams.errors.TaskCorruptedException;
  import org.apache.kafka.streams.processor.StateStore;
  import org.apache.kafka.streams.processor.TaskId;
  
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  
  public interface Task {
  
      // this must be negative to distinguish a running active task from other kinds of tasks
      // which may be caught up to the same offsets
      long LATEST_OFFSET = -2L;
  
      /*
       * <pre>
       *                 +---------------+
       *          +----- |  Created (0)  | <---------+
       *          |      +-------+-------+           |
       *          |              |                   |
       *          |              v                   |
       *          |      +-------+-------+           |
       *          +----- | Restoring (1) | <---+     |
       *          |      +-------+-------+     |     |
       *          |              |             |     |
       *          |              v             |     |
       *          |      +-------+-------+     |     |
       *          |      |  Running (2)  |     |     |
       *          |      +-------+-------+     |     |
       *          |              |             |     |
       *          |              v             |     |
       *          |     +--------+-------+     |     |
       *          +---> |  Suspended (3) | ----+     |    //TODO Suspended(3) could be removed after we've stable on KIP-429
       *                +--------+-------+           |
       *                         |                   |
       *                         v                   |
       *                +--------+-------+           |
       *                |   Closed (4)   | ----------+
       *                +----------------+
       * </pre>
       */
      enum State {
          CREATED(1, 3),            // 0
          RESTORING(2, 3),          // 1
          RUNNING(3),               // 2
          SUSPENDED(1, 4),          // 3
          CLOSED(0);                // 4, we allow CLOSED to transit to CREATED to handle corrupted tasks
  
          private final Set<Integer> validTransitions = new HashSet<>();
  
          State(final Integer... validTransitions) {
              this.validTransitions.addAll(Arrays.asList(validTransitions));
          }
  
          public boolean isValidTransition(final State newState) {
              return validTransitions.contains(newState.ordinal());
          }
      }
  
      enum TaskType {
          ACTIVE("ACTIVE"),
  
          STANDBY("STANDBY"),
  
          GLOBAL("GLOBAL");
  
          public final String name;
  
          TaskType(final String name) {
              this.name = name;
          }
      }
  
  
      // idempotent life-cycle methods
  
      /**
       * @throws TaskCorruptedException if the state cannot be reused (with EOS) and needs to be reset
       * @throws LockException    could happen when multi-threads within the single instance, could retry
       * @throws StreamsException fatal error, should close the thread
       */
      void initializeIfNeeded();
  
      default void addPartitionsForOffsetReset(final Set<TopicPartition> partitionsForOffsetReset) {
          throw new UnsupportedOperationException();
      }
  
      /**
       * @throws StreamsException fatal error, should close the thread
       */
      void completeRestoration(final java.util.function.Consumer<Set<TopicPartition>> offsetResetter);
  
      void suspend();
  
      /**
       * @throws StreamsException fatal error, should close the thread
       */
      void resume();
  
      /**
       * Must be idempotent.
       */
      void closeDirty();
  
      /**
       * Must be idempotent.
       */
      void closeClean();
  
  
      // non-idempotent life-cycle methods
  
      /**
       * Updates input partitions and topology after rebalance
       */
      void updateInputPartitions(final Set<TopicPartition> topicPartitions, final Map<String, List<String>> allTopologyNodesToSourceTopics);
  
      /**
       * @param enforceCheckpoint if true the task would always execute the checkpoint;
       *                          otherwise it may skip if the state has not advanced much
       */
      void maybeCheckpoint(final boolean enforceCheckpoint);
  
      void markChangelogAsCorrupted(final Collection<TopicPartition> partitions);
  
      /**
       * Revive a closed task to a created one; should never throw an exception
       */
      void revive();
  
      /**
       * Close the task except the state, so that the states can be later recycled
       */
      void prepareRecycle();
  
      /**
       * Resumes polling in the main consumer for all partitions for which
       * the corresponding record queues have capacity (again).
       */
      void resumePollingForPartitionsWithAvailableSpace();
  
      /**
       * Fetches up-to-date lag information from the consumer.
       */
      void updateLags();
  
      // runtime methods (using in RUNNING state)
  
      void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records);
  
      default void updateNextOffsets(final TopicPartition partition, final OffsetAndMetadata offsetAndMetadata) {
      }
  
      default boolean process(final long wallClockTime) {
          return false;
      }
  
      default void recordProcessBatchTime(final long processBatchTime) {
      }
  
      default void recordProcessTimeRatioAndBufferSize(final long allTaskProcessMs, final long now) {
      }
  
      default boolean maybePunctuateStreamTime() {
          return false;
      }
  
      default boolean maybePunctuateSystemTime() {
          return false;
      }
  
      /**
       * @throws StreamsException fatal error, should close the thread
       */
      Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean);
  
      void postCommit(boolean enforceCheckpoint);
  
      default Map<TopicPartition, Long> purgeableOffsets() {
          return Collections.emptyMap();
      }
  
      /**
       * @throws StreamsException if {@code currentWallClockMs > task-timeout-deadline}
       */
      void maybeInitTaskTimeoutOrThrow(final long currentWallClockMs,
                                       final Exception cause);
  
      void clearTaskTimeout();
  
      void recordRestoration(final Time time, final long numRecords, final boolean initRemaining);
  
      // task status inquiry
  
      TaskId id();
  
      boolean isActive();
  
      Set<TopicPartition> inputPartitions();
  
      /**
       * @return any changelog partitions associated with this task
       */
      Set<TopicPartition> changelogPartitions();
  
      State state();
  
      ProcessorStateManager stateManager();
  
      default boolean needsInitializationOrRestoration() {
          return state() == State.CREATED || state() == State.RESTORING;
      }
  
      boolean commitNeeded();
  
      default boolean commitRequested() {
          return false;
      }
  
  
      // IQ related methods
  
      StateStore store(final String name);
  
      /**
       * @return the offsets of all the changelog partitions associated with this task,
       * indicating the current positions of the logged state stores of the task.
       */
      Map<TopicPartition, Long> changelogOffsets();
  
      /**
       * @return the offsets that each TopicPartition has committed so far in this task,
       * indicating how far the processing has committed
       */
      Map<TopicPartition, Long> committedOffsets();
  
      /**
       * @return the highest offsets that each TopicPartition has seen so far in this task
       */
      Map<TopicPartition, Long> highWaterMark();
  
      /**
       * @return This returns the time the task started idling. If it is not idling it returns empty.
       */
      Optional<Long> timeCurrentIdlingStarted();
  
  }
  