/**
 * 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;
}

public interface StreamsMetadata {

    /**
     * The value of {@link StreamsConfig#APPLICATION_SERVER_CONFIG} configured for the streams
     * instance, which is typically host/port
     *
     * @return {@link HostInfo} corresponding to the streams instance
     */
    HostInfo hostInfo();

    /**
     * State stores owned by the instance as an active replica
     *
     * @return set of active state store names
     */
    Set<String> stateStoreNames();

    

/**
     * Topic partitions Findconsumed allby currentlythe runninginstance {@codeas KafkaStreams}an instancesactive (potentiallyreplica
 remotely) that use the same*
     * {@link StreamsConfig#APPLICATION_ID_CONFIG application ID} as this instance (i.e., all instances that belong to@return set of active topic partitions
     */
    Set<TopicPartition> topicPartitions();

    /**
 * the same Kafka Streams* application(Source) Topic andpartitions returnfor {@linkwhich StreamsMetadata}the forinstance eachacts discoveredas instancestandby.
     * <p>
     * Note:@return thisset isof astandby pointtopic inpartitions
 time view and it may*/
 change due to partition reassignment.Set<TopicPartition> standbyTopicPartitions();

     /**
     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances of this applicationState stores owned by the instance as a standby replica
     *
     * @return @deprecatedset since 3.0.0 use {@link KafkaStreams#allRunningMetadata}of standby state store names
     */
    @DeprecatedSet<String> standbyStateStoreNames();

    public Collection<org.apache.kafka.streams.state.StreamsMetadata> allMetadata() {
        validateIsRunningOrRebalancing();/**
     * This method is equivalent to call {@code StreamsMetadata.hostInfo().host();}
     */
    returnString streamsMetadataState.getAllMetadata().stream().map(streamsMetadata ->host();

    /**
     * This method is equivalent to call new{@code orgStreamsMetadata.apache.kafka.streams.state.StreamsMetadata(streamsMetadata.hostInfo(),hostInfo().port();}
     */
    int port();

    /**
     * Compares the specified object streamsMetadata.stateStoreNames(),
           with this StreamsMetadata. Returns {@code true} if and only if the specified object is
     * also a StreamsMetadata and for both {@code streamsMetadata.topicPartitionshostInfo(),
} are equal, and {@code stateStoreNames()}, {@code topicPartitions()},
     * {@code standbyStateStoreNames()}, and {@code standbyTopicPartitions()} contain the same elements.
   streamsMetadata.standbyStateStoreNames(),  */
    boolean equals(Object o);

    /**
     * Returns the hash code value for this  streamsMetadata.standbyTopicPartitions()))
                .collect(Collectors.toSet());
    }

    /**TaskMetadata. The hash code of a list is defined to be the result of the following calculation:
     * <pre>
     * Find{@code
  all currently running * Objects.hash(hostInfo(), stateStoreNames(), topicPartitions(), standbyStateStoreNames(), standbyTopicPartitions());
     * }
     */
    int hashCode();

}

/**
 * Represents the state of an instance (process) in a {@link KafkaStreams} application.
 * It contains the user supplied {@link HostInfo} that can be used by developers to build
 * APIs and services to connect to other instances, the Set of state stores available on
 * the instance and the Set of {@link TopicPartition}s available on the instance.
 * NOTE: This is a point in time view. It may change when rebalances happen.
 * @deprecated since 3.0.0 use {@link org.apache.kafka.streams.StreamsMetadata}
 */
@Deprecated
public class StreamsMetadata {
...
}

{@code KafkaStreams} instances (potentially remotely) that use the same
     * {@link StreamsConfig#APPLICATION_ID_CONFIG application ID} as this instance (i.e., all instances that belong to
     * the same Kafka Streams application) and return {@link StreamsMetadata} for each discovered instance.
     * <p>
     * Note: this is a point in time view and it may change due to partition reassignment.
     *
     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances of this application
     */
    public Collection<StreamsMetadata> allRunningMetadata() {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.getAllMetadata();
    }

    /**
     * Find all currently running {@code KafkaStreams} instances (potentially remotely) that
 use the   * <ul>same
     *   <li>use the same {@link StreamsConfig#APPLICATION{@link StreamsConfig#APPLICATION_ID_CONFIG application ID} as this instance (i.e., all
     *       instances that belong to the same Kafka Streams application)</li>
     *   <li>and that contain a {@link StateStore} with the givensame {@code storeName}</li>
     * </ul>
     * Kafka Streams application) and return {@link StreamsMetadata} for each discovered instance.
     * <p>
     * Note: this is a point in time view and it may change due to partition reassignment.
     *
     * @param storeName the {@code storeName} to find metadata for
     * @return {@return {@link StreamsMetadata} for each {@code KafkaStreams} instances with the provide {@code storeName} of
     * this application
     * @deprecated since 3.0.0 use {@link KafkaStreams#allMetadataForGivenStoreKafkaStreams#allRunningMetadata} instead
     */
    @Deprecated
    public Collection<org.apache.kafka.streams.state.StreamsMetadata> allMetadataForStoreallMetadata(final String storeName) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.getAllMetadataForStoregetAllMetadata(storeName).stream().map(streamsMetadata ->
                new org.apache.kafka.streams.state.StreamsMetadata(streamsMetadata.hostInfo(),
                        streamsMetadata.stateStoreNames(),
                        streamsMetadata.topicPartitions(),
                        streamsMetadata.standbyStateStoreNames(),
                        streamsMetadata.standbyTopicPartitions()))
                .collect(Collectors.toSet());
    }

    /**
     * Find all currently running {@code KafkaStreams} instances (potentially remotely) that
 use the   * <ul>same
     *   <li>use the same {@link StreamsConfig#APPLICATION_ID_CONFIG application ID} as this instance (i.e., all
     *       instances that belong to the same Kafka Streams application)</li>
     *   <li>and that contain a {@link StateStore} with the given {@code storeName}</li>
     * </ul>
     * and return {@link StreamsMetadata} for each discovered instance.
     * <p>
     * Note: this is a point in time view and it may change due to partition reassignment.
     *
     * @param storeName the {@code storeName} to find metadata for
     * @return {@link StreamsMetadata} for each {@code KafkaStreams} instances with the provide {@code storeName} of
     * this application
     */
    public Collection<StreamsMetadata> allMetadataForGivenStore(final String storeName) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.getAllMetadataForStore(storeName);
    }