/**
 * Represents the state of a single task running within a {@link KafkaStreams} application.
 */
public interface TaskMetadata {


    /**
     * @return the basic task metadata such as subtopology and partition id
     */
    TaskId taskId();

    /**
     * This function will return a set of the current TopicPartitions
     */
    Set<TopicPartition> topicPartitions();

    /**
     * This function will return a map of TopicPartitions and the highest committed offset seen so far
     */
    Map<TopicPartition, Long> committedOffsets();

    /**
     * This function will return a map of TopicPartitions and the highest offset seen so far in the Topic
     */
    Map<TopicPartition, Long> endOffsets();

    /**
     * This function will return the time task idling started, if the task is not currently idling it will return empty
     */
    Optional<Long> timeCurrentIdlingStarted();

	/**
     * Compares the specified object with this TaskMetadata. Returns {@code true} if and only if the specified object is
     * also a TaskMetadata and both {@code taskId()} and {@code topicPartitions()} are equal.
     */
    boolean equals(final Object o);

    /**
     * Returns the hash code value for this TaskMetadata. The hash code of a list is defined to be the result of the following calculation:
     * <pre>
     * {@code
     * Objects.hash(taskId(), topicPartitions());
     * }
     * </pre>
     */
    int hashCode();
}

/**
 * Represents the state of a single task running within a {@link KafkaStreams} application.
 * @deprecated since 3.0, use {@link org.apache.kafka.streams.TaskMetadata} instead
 */
@Deprecated
public class TaskMetadata {
...
}

// This method will be removed in this KIP
/**
 * @return the basic task metadata such as subtopology and partition id
*/
public TaskId getTaskId() {
   return taskId;
}

/**
 * Represents the state of a single thread running within a {@link KafkaStreams} application.
 * @deprecated since 3.0 use {@link org.apache.kafka.streams.ThreadMetadata} instead
 */
@Deprecated
public class ThreadMetadata {
...
}

/**
 * Represents the state of a single thread running within a {@link KafkaStreams} application.
 */
public interface ThreadMetadata {


    /**
     * @return the state of the Thread
     */
    String threadState();

    /**
     * @return the name of the Thread
     */
    String threadName();

    /**
     * This function will return the set of the {@link TaskMetadata} for the current active tasks
     */
    Set<TaskMetadata> activeTasks();

    /**
     * This function will return the set of the {@link TaskMetadata} for the current standby tasks
     */
    Set<TaskMetadata> standbyTasks();

    /**
     * @return the consumer Client Id
     */
    String consumerClientId();

    /**
     * @return the restore consumer Client Id
     */
    String restoreConsumerClientId();

    /**
     * This function will return the set of Client Ids for the producers
     */
    Set<String> producerClientIds();

    /**
     * @return the admin Client Id
     */
    String adminClientId();

    /**
     * Compares the specified object with this ThreadMetadata. Returns {@code true} if and only if the specified object is
     * also a TaskMetadata and both {@code threadName()} are equal, {@code threadState()} are equal, {@code activeTasks()} contain the same
     * elements, {@code standbyTasks()} contain the same elements, {@code mainConsumerClientId()} are equal, {@code restoreConsumerClientId()}
     * are equal, {@code producerClientIds()} are equal, {@code producerClientIds} contain the same elements, and {@code adminClientId()} are equal.
     */
    boolean equals(Object o);

    /**
     * Returns the hash code value for this TaskMetadata. The hash code of a list is defined to be the result of the following calculation:
     * <pre>
     * {@code
     * Objects.hash(
     *             threadName,
     *             threadState,
     *             activeTasks,
     *             standbyTasks,
     *             mainConsumerClientId,
     *             restoreConsumerClientId,
     *             producerClientIds,
     *             adminClientId);
     * </pre>
     */
    int hashCode();
}

	/**
     * Returns runtime information about the local threads of this {@link KafkaStreams} instance.
     *
     * @return the set of {@link org.apache.kafka.streams.processor.ThreadMetadata}.
     * @deprecated since 3.0 use {@link #threadsMetadata()}
     */
    @Deprecated
    public Set<org.apache.kafka.streams.processor.ThreadMetadata> localThreadsMetadata() {
        return threadsMetadata().stream().map(threadMetadata -> new org.apache.kafka.streams.processor.ThreadMetadata(
                threadMetadata.threadName(),
                threadMetadata.threadState(),
                threadMetadata.consumerClientId(),
                threadMetadata.restoreConsumerClientId(),
                threadMetadata.producerClientIds(),
                threadMetadata.adminClientId(),
                threadMetadata.activeTasks().stream().map(taskMetadata -> new TaskMetadata(
                        taskMetadata.taskId().toString(),
                        taskMetadata.topicPartitions(),
                        taskMetadata.committedOffsets(),
                        taskMetadata.endOffsets(),
                        taskMetadata.timeCurrentIdlingStarted())
                ).collect(Collectors.toSet()),
                threadMetadata.standbyTasks().stream().map(taskMetadata -> new TaskMetadata(
                        taskMetadata.taskId().toString(),
                        taskMetadata.topicPartitions(),
                        taskMetadata.committedOffsets(),
                        taskMetadata.endOffsets(),
                        taskMetadata.timeCurrentIdlingStarted())
                ).collect(Collectors.toSet())))
                .collect(Collectors.toSet());
    }

    /**
     * Returns runtime information about the local threads of this {@link KafkaStreams} instance.
     *
     * @return the set of {@link ThreadMetadata}.
     */
    public Set<ThreadMetadata> threadsMetadata() {
        final Set<ThreadMetadata> threadMetadata = new HashSet<>();
        processStreamThread(thread -> {
            synchronized (thread.getStateLock()) {
                if (thread.state() != StreamThread.State.DEAD) {
                    threadMetadata.add(thread.getThreadMetadata());
                }
            }
        });
        return threadMetadata;
    }