You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2024/08/23 03:18:37 UTC
(kafka) branch trunk updated: KAFKA-16329: Remove deprecated IQ metadata classes in Kafka Streams (#16762)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8ceb05d203f KAFKA-16329: Remove deprecated IQ metadata classes in Kafka Streams (#16762)
8ceb05d203f is described below
commit 8ceb05d203f48a0312019559b433cfc4b92c6698
Author: Murali Basani <mu...@aiven.io>
AuthorDate: Fri Aug 23 05:18:30 2024 +0200
KAFKA-16329: Remove deprecated IQ metadata classes in Kafka Streams (#16762)
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../org/apache/kafka/streams/KafkaStreams.java | 83 -----------
.../kafka/streams/processor/TaskMetadata.java | 117 ----------------
.../kafka/streams/processor/ThreadMetadata.java | 145 -------------------
.../kafka/streams/state/StreamsMetadata.java | 156 ---------------------
4 files changed, 501 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 74c3568eb64..f3a1126a24a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -107,7 +107,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;
-import java.util.stream.Collectors;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
import static org.apache.kafka.streams.internals.ApiUtils.prepareMillisCheckFailMsgPrefix;
@@ -1675,28 +1674,6 @@ public class KafkaStreams implements AutoCloseable {
stateDirectory.clean();
}
- /**
- * Find all currently running {@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
- * @deprecated since 3.0.0 use {@link KafkaStreams#metadataForAllStreamsClients}
- */
- @Deprecated
- public Collection<org.apache.kafka.streams.state.StreamsMetadata> allMetadata() {
- validateIsRunningOrRebalancing();
- return streamsMetadataState.getAllMetadata().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 same
* {@link StreamsConfig#APPLICATION_ID_CONFIG application ID} as this instance (i.e., all instances that belong to
@@ -1711,34 +1688,6 @@ public class KafkaStreams implements AutoCloseable {
return streamsMetadataState.getAllMetadata();
}
- /**
- * Find all currently running {@code KafkaStreams} instances (potentially remotely) that
- * <ul>
- * <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
- * @deprecated since 3.0.0 use {@link KafkaStreams#streamsMetadataForStore} instead
- */
- @Deprecated
- public Collection<org.apache.kafka.streams.state.StreamsMetadata> allMetadataForStore(final String storeName) {
- validateIsRunningOrRebalancing();
- return streamsMetadataState.getAllMetadataForStore(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
* <ul>
@@ -2040,38 +1989,6 @@ public class KafkaStreams implements AutoCloseable {
throw new StreamsException(errorMessage.get(), cause);
}
- /**
- * 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 #metadataForLocalThreads()}
- */
- @Deprecated
- public Set<org.apache.kafka.streams.processor.ThreadMetadata> localThreadsMetadata() {
- return metadataForLocalThreads().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 org.apache.kafka.streams.processor.TaskMetadata(
- taskMetadata.taskId().toString(),
- taskMetadata.topicPartitions(),
- taskMetadata.committedOffsets(),
- taskMetadata.endOffsets(),
- taskMetadata.timeCurrentIdlingStarted())
- ).collect(Collectors.toSet()),
- threadMetadata.standbyTasks().stream().map(taskMetadata -> new org.apache.kafka.streams.processor.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.
*
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java b/streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java
deleted file mode 100644
index b9f5d915abc..00000000000
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TaskMetadata.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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;
-
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.streams.KafkaStreams;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-
-/**
- * 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 {
-
- private final String taskId;
-
- private final Set<TopicPartition> topicPartitions;
-
- private final Map<TopicPartition, Long> committedOffsets;
-
- private final Map<TopicPartition, Long> endOffsets;
-
- private final Optional<Long> timeCurrentIdlingStarted;
-
- public TaskMetadata(final String taskId,
- final Set<TopicPartition> topicPartitions,
- final Map<TopicPartition, Long> committedOffsets,
- final Map<TopicPartition, Long> endOffsets,
- final Optional<Long> timeCurrentIdlingStarted) {
- this.taskId = taskId;
- this.topicPartitions = Collections.unmodifiableSet(topicPartitions);
- this.committedOffsets = Collections.unmodifiableMap(committedOffsets);
- this.endOffsets = Collections.unmodifiableMap(endOffsets);
- this.timeCurrentIdlingStarted = timeCurrentIdlingStarted;
- }
-
- /**
- * @return the basic task metadata such as subtopology and partition id
- */
- public String taskId() {
- return taskId;
- }
-
- public Set<TopicPartition> topicPartitions() {
- return topicPartitions;
- }
-
- /**
- * This function will return a map of TopicPartitions and the highest committed offset seen so far
- */
- public Map<TopicPartition, Long> committedOffsets() {
- return committedOffsets;
- }
-
- /**
- * This function will return a map of TopicPartitions and the highest offset seen so far in the Topic
- */
- public Map<TopicPartition, Long> endOffsets() {
- return endOffsets;
- }
-
- /**
- * This function will return the time task idling started, if the task is not currently idling it will return empty
- */
- public Optional<Long> timeCurrentIdlingStarted() {
- return timeCurrentIdlingStarted;
- }
-
- @Override
- public boolean equals(final Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- final TaskMetadata that = (TaskMetadata) o;
- return Objects.equals(taskId, that.taskId) &&
- Objects.equals(topicPartitions, that.topicPartitions);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(taskId, topicPartitions);
- }
-
- @Override
- public String toString() {
- return "TaskMetadata{" +
- "taskId=" + taskId +
- ", topicPartitions=" + topicPartitions +
- ", committedOffsets=" + committedOffsets +
- ", endOffsets=" + endOffsets +
- ", timeCurrentIdlingStarted=" + timeCurrentIdlingStarted +
- '}';
- }
-}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ThreadMetadata.java b/streams/src/main/java/org/apache/kafka/streams/processor/ThreadMetadata.java
deleted file mode 100644
index 68ac6633b76..00000000000
--- a/streams/src/main/java/org/apache/kafka/streams/processor/ThreadMetadata.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
- * 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;
-
-import org.apache.kafka.streams.KafkaStreams;
-
-import java.util.Collections;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * 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 {
-
- private final String threadName;
-
- private final String threadState;
-
- private final Set<TaskMetadata> activeTasks;
-
- private final Set<TaskMetadata> standbyTasks;
-
- private final String mainConsumerClientId;
-
- private final String restoreConsumerClientId;
-
- private final Set<String> producerClientIds;
-
- // the admin client should be shared among all threads, so the client id should be the same;
- // we keep it at the thread-level for user's convenience and possible extensions in the future
- private final String adminClientId;
-
- public ThreadMetadata(final String threadName,
- final String threadState,
- final String mainConsumerClientId,
- final String restoreConsumerClientId,
- final Set<String> producerClientIds,
- final String adminClientId,
- final Set<org.apache.kafka.streams.processor.TaskMetadata> activeTasks,
- final Set<org.apache.kafka.streams.processor.TaskMetadata> standbyTasks) {
- this.mainConsumerClientId = mainConsumerClientId;
- this.restoreConsumerClientId = restoreConsumerClientId;
- this.producerClientIds = producerClientIds;
- this.adminClientId = adminClientId;
- this.threadName = threadName;
- this.threadState = threadState;
- this.activeTasks = Collections.unmodifiableSet(activeTasks);
- this.standbyTasks = Collections.unmodifiableSet(standbyTasks);
- }
-
- public String threadState() {
- return threadState;
- }
-
- public String threadName() {
- return threadName;
- }
-
- public Set<org.apache.kafka.streams.processor.TaskMetadata> activeTasks() {
- return activeTasks;
- }
-
- public Set<org.apache.kafka.streams.processor.TaskMetadata> standbyTasks() {
- return standbyTasks;
- }
-
- public String consumerClientId() {
- return mainConsumerClientId;
- }
-
- public String restoreConsumerClientId() {
- return restoreConsumerClientId;
- }
-
- public Set<String> producerClientIds() {
- return producerClientIds;
- }
-
- public String adminClientId() {
- return adminClientId;
- }
-
- @Override
- public boolean equals(final Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- final ThreadMetadata that = (ThreadMetadata) o;
- return Objects.equals(threadName, that.threadName) &&
- Objects.equals(threadState, that.threadState) &&
- Objects.equals(activeTasks, that.activeTasks) &&
- Objects.equals(standbyTasks, that.standbyTasks) &&
- mainConsumerClientId.equals(that.mainConsumerClientId) &&
- restoreConsumerClientId.equals(that.restoreConsumerClientId) &&
- Objects.equals(producerClientIds, that.producerClientIds) &&
- adminClientId.equals(that.adminClientId);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(
- threadName,
- threadState,
- activeTasks,
- standbyTasks,
- mainConsumerClientId,
- restoreConsumerClientId,
- producerClientIds,
- adminClientId);
- }
-
- @Override
- public String toString() {
- return "ThreadMetadata{" +
- "threadName=" + threadName +
- ", threadState=" + threadState +
- ", activeTasks=" + activeTasks +
- ", standbyTasks=" + standbyTasks +
- ", consumerClientId=" + mainConsumerClientId +
- ", restoreConsumerClientId=" + restoreConsumerClientId +
- ", producerClientIds=" + producerClientIds +
- ", adminClientId=" + adminClientId +
- '}';
- }
-}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java b/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java
deleted file mode 100644
index d596d6a7e38..00000000000
--- a/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/*
- * 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.state;
-
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.streams.KafkaStreams;
-
-import java.util.Collections;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * 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 {
- /**
- * Sentinel to indicate that the StreamsMetadata is currently unavailable. This can occur during rebalance
- * operations.
- */
- public static final StreamsMetadata NOT_AVAILABLE = new StreamsMetadata(HostInfo.unavailable(),
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet());
-
- private final HostInfo hostInfo;
-
- private final Set<String> stateStoreNames;
-
- private final Set<TopicPartition> topicPartitions;
-
- private final Set<String> standbyStateStoreNames;
-
- private final Set<TopicPartition> standbyTopicPartitions;
-
- public StreamsMetadata(final HostInfo hostInfo,
- final Set<String> stateStoreNames,
- final Set<TopicPartition> topicPartitions,
- final Set<String> standbyStateStoreNames,
- final Set<TopicPartition> standbyTopicPartitions) {
-
- this.hostInfo = hostInfo;
- this.stateStoreNames = stateStoreNames;
- this.topicPartitions = topicPartitions;
- this.standbyTopicPartitions = standbyTopicPartitions;
- this.standbyStateStoreNames = standbyStateStoreNames;
- }
-
- /**
- * The value of {@link org.apache.kafka.streams.StreamsConfig#APPLICATION_SERVER_CONFIG} configured for the streams
- * instance, which is typically host/port
- *
- * @return {@link HostInfo} corresponding to the streams instance
- */
- public HostInfo hostInfo() {
- return hostInfo;
- }
-
- /**
- * State stores owned by the instance as an active replica
- *
- * @return set of active state store names
- */
- public Set<String> stateStoreNames() {
- return Collections.unmodifiableSet(stateStoreNames);
- }
-
- /**
- * Topic partitions consumed by the instance as an active replica
- *
- * @return set of active topic partitions
- */
- public Set<TopicPartition> topicPartitions() {
- return Collections.unmodifiableSet(topicPartitions);
- }
-
- /**
- * (Source) Topic partitions for which the instance acts as standby.
- *
- * @return set of standby topic partitions
- */
- public Set<TopicPartition> standbyTopicPartitions() {
- return Collections.unmodifiableSet(standbyTopicPartitions);
- }
-
- /**
- * State stores owned by the instance as a standby replica
- *
- * @return set of standby state store names
- */
- public Set<String> standbyStateStoreNames() {
- return Collections.unmodifiableSet(standbyStateStoreNames);
- }
-
- public String host() {
- return hostInfo.host();
- }
-
- @SuppressWarnings("unused")
- public int port() {
- return hostInfo.port();
- }
-
- @Override
- public boolean equals(final Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- final StreamsMetadata that = (StreamsMetadata) o;
- return Objects.equals(hostInfo, that.hostInfo)
- && Objects.equals(stateStoreNames, that.stateStoreNames)
- && Objects.equals(topicPartitions, that.topicPartitions)
- && Objects.equals(standbyStateStoreNames, that.standbyStateStoreNames)
- && Objects.equals(standbyTopicPartitions, that.standbyTopicPartitions);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(hostInfo, stateStoreNames, topicPartitions, standbyStateStoreNames, standbyTopicPartitions);
- }
-
- @Override
- public String toString() {
- return "StreamsMetadata {" +
- "hostInfo=" + hostInfo +
- ", stateStoreNames=" + stateStoreNames +
- ", topicPartitions=" + topicPartitions +
- ", standbyStateStoreNames=" + standbyStateStoreNames +
- ", standbyTopicPartitions=" + standbyTopicPartitions +
- '}';
- }
-}