You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sjhajharia (via GitHub)" <gi...@apache.org> on 2024/11/12 10:05:25 UTC
[PR] [DRAFT] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
sjhajharia opened a new pull request, #17775:
URL: https://github.com/apache/kafka/pull/17775
THIS IS A DRAFT PR.
This PR aims at adding the rough code for the first part of KAFKA-16720.
- Adding the 3 java classes
- Changes on the ShareCoordinator to handle the ReadShareGroupStateSummaryRequest
Pending items:
- Group Coordinator changes
- Default State Persistor changes
- Admin Client updates
- Tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1908544013
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -175,6 +189,8 @@ public GroupCoordinatorService build() {
throw new IllegalArgumentException("GroupCoordinatorMetrics must be set.");
if (groupConfigManager == null)
throw new IllegalArgumentException("GroupConfigManager must be set.");
+ if (persister == null)
+ throw new IllegalArgumentException("Persister must be set.");
Review Comment:
Right now yes.
We may remove this condition here though. I will check that again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DRAFT] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1837924338
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -17,10 +17,7 @@
package org.apache.kafka.coordinator.share;
-import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
-import org.apache.kafka.common.message.ReadShareGroupStateResponseData;
-import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
-import org.apache.kafka.common.message.WriteShareGroupStateResponseData;
+import org.apache.kafka.common.message.*;
Review Comment:
will fix in the next commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1908542698
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java:
##########
@@ -210,6 +212,18 @@ CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGrou
List<String> groupIds
);
+ /**
+ * List share group Offsets.
+ *
+ * @param context The coordinator request context.
+ * @param request The ReadShareGroupStateSummaryRequest data.
+ *
+ * @return A future yielding the results or an exception.
+ */
+ CompletableFuture<List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult>> listShareGroupOffsets(
Review Comment:
Sorry, I don't follow that. Could you pls elaborate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia closed pull request #17775: KAFKA 16720 : AdminClient support for ListShareGroupOffsets
URL: https://github.com/apache/kafka/pull/17775
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1904599414
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -244,6 +261,11 @@ public GroupCoordinatorService build() {
*/
private final GroupConfigManager groupConfigManager;
+ /**
+ * The Persister to persist the state of GC
+ */
+ private final Persister persister;
Review Comment:
I don't see why the persister is necessary here.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(groupIds.size());
+ groupIds.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? new ArrayList<>(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public Collection<RequestAndKeys<CoordinatorKey>> buildRequest(int brokerId, Set<CoordinatorKey> groupIds) {
+ validateKeys(groupIds);
+
+ // When the OffsetFetchRequest fails with NoBatchedOffsetFetchRequestException, we completely disable
+ // the batching end-to-end, including the FindCoordinatorRequest.
+ if (lookupStrategy.batch()) {
+ return Collections.singletonList(new RequestAndKeys<>(buildBatchedRequest(groupIds), groupIds));
+ } else {
+ return groupIds.stream().map(groupId -> {
+ Set<CoordinatorKey> keys = Collections.singleton(groupId);
+ return new RequestAndKeys<>(buildBatchedRequest(keys), keys);
+ }).collect(Collectors.toList());
+ }
+ }
+
+ @Override
+ public ApiResult<CoordinatorKey, Map<TopicPartition, Long>> handleResponse(
+ Node coordinator,
+ Set<CoordinatorKey> groupIds,
+ AbstractResponse abstractResponse
+ ) {
+ validateKeys(groupIds);
+
+ final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
+
+ Map<CoordinatorKey, Map<TopicPartition, Long>> completed = new HashMap<>();
+ Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ List<CoordinatorKey> unmapped = new ArrayList<>();
+ for (CoordinatorKey coordinatorKey : groupIds) {
+ String group = coordinatorKey.idValue;
+ if (response.groupHasError(group)) {
+ handleGroupError(CoordinatorKey.byGroupId(group), response.groupLevelError(group), failed, unmapped);
+ } else {
+ final Map<TopicPartition, Long> groupOffsetsListing = new HashMap<>();
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = response.partitionDataMap(group);
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> partitionEntry : responseData.entrySet()) {
+ final TopicPartition topicPartition = partitionEntry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = partitionEntry.getValue();
+ final Errors error = partitionData.error;
+
+ if (error == Errors.NONE) {
+ final long offset = partitionData.offset;
+ // Negative offset indicates that the group has no committed offset for this partition
+ if (offset < 0) {
+ groupOffsetsListing.put(topicPartition, null);
+ } else {
+ groupOffsetsListing.put(topicPartition, offset);
+ }
+ } else {
+ log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
+ }
+ }
+ completed.put(CoordinatorKey.byGroupId(group), groupOffsetsListing);
+ }
+ }
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private void handleGroupError(
+ CoordinatorKey groupId,
+ Errors error,
+ Map<CoordinatorKey, Throwable> failed,
+ List<CoordinatorKey> groupsToUnmap
+ ) {
+ switch (error) {
+ case GROUP_AUTHORIZATION_FAILED:
+ case UNKNOWN_MEMBER_ID:
+ case STALE_MEMBER_EPOCH:
+ log.debug("`OffsetFetch` request for group id {} failed due to error {}", groupId.idValue, error);
Review Comment:
The remnants of `OffsetFetch` are still evident here. I don't think you need handling for `UNKNOWN_MEMBER_ID` or `STALE_MEMBER_EPOCH`. You do need `GROUP_ID_NOT_FOUND`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
Review Comment:
Agreed. It seems to me that `extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>>` is better.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(groupIds.size());
+ groupIds.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? new ArrayList<>(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public Collection<RequestAndKeys<CoordinatorKey>> buildRequest(int brokerId, Set<CoordinatorKey> groupIds) {
Review Comment:
I would model this on `DescribeShareGroupsHandler`. We are going to add a bunch more handlers in AK 4.1 to fill out the rest of the admin support for KIP-932. There should be a lot of similarity between these handlers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #17775:
URL: https://github.com/apache/kafka/pull/17775#issuecomment-2580048378
Hey @AndrewJSchofield / @apoorvmittal10
Could I pls get another review on the PR.
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DRAFT] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1844139747
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * The result of the {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsResult {
+
+ private final Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures;
+
+ public ListShareGroupOffsetsResult(Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures) {
+ this.futures = futures;
+ }
+
+ /**
+ * Return a future which yields all Map<String, Map<TopicPartition, Long> objects, if requests for all the groups succeed.
+ */
+ public KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
+ nil -> {
+ Map<String, Map<TopicPartition, Long>> offsets = new HashMap<>(futures.size());
+ futures.forEach((key, future) -> {
+ try {
+ offsets.put(key, future.get());
+ } catch (InterruptedException | ExecutionException e) {
+ // This should be unreachable, since the KafkaFuture#allOf already ensured
+ // that all the futures completed successfully.
+ throw new RuntimeException(e);
+ }
+ });
+ return offsets;
+ });
+ }
+
+ /**
+ * Return a future which yields a map of topic partitions to offsets for the specified group.
+ */
+ public KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
+ KafkaFutureImpl<Map<TopicPartition, Long>> future = new KafkaFutureImpl<>();
+ if (futures.containsKey(groupId))
+ return futures.get(groupId);
+ else
+ future.completeExceptionally(new IllegalArgumentException("Group ID not found: " + groupId));
Review Comment:
I would make this throw `IllegalArgumentException` rather than complete the future exceptionally. See `ListConsumerGroupOffsetsResult`.partitionsToOffsetAndMetadata`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * The result of the {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsResult {
+
+ private final Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures;
+
+ public ListShareGroupOffsetsResult(Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures) {
Review Comment:
This method should be package-private, not public. Otherwise it becomes part of the public interface and will also end up in the javadoc.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -478,6 +481,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readState method in ShareCoordinatorShard expects a single key in the request. Hence, we will
+ // be looping over the keys below and constructing new ReadShareGroupStateRequestData objects to pass
+ // onto the shard method.
+
+ request.topics().forEach(topicData -> {
+ Uuid topicId = topicData.topicId();
+ topicData.partitions().forEach(partitionData -> {
+ // Request object containing information of a single topic partition
+ ReadShareGroupStateSummaryRequestData requestForCurrentPartition = new ReadShareGroupStateSummaryRequestData()
+ .setGroupId(groupId)
+ .setTopics(Collections.singletonList(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
Review Comment:
List.of
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -478,6 +481,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readState method in ShareCoordinatorShard expects a single key in the request. Hence, we will
Review Comment:
"summary"
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +733,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
+ }
+
+ @Override
+ protected AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> requestBuilder() {
+ throw new RuntimeException("Read Summary requests are batchable, hence individual requests not needed.");
+ }
+
+ @Override
+ protected boolean isResponseForRequest(ClientResponse response) {
+ return response.requestHeader().apiKey() == ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY;
+ }
+
+ @Override
+ protected void handleRequestResponse(ClientResponse response) {
+ log.debug("Read state summary response received - {}", response);
+ readStateBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateResult : combinedResponse.data().results()) {
+ if (readStateResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateBackoff.resetAttempts();
+ ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult result = ReadShareGroupStateSummaryResponse.toResponseReadStateSummaryResult(
+ partitionKey().topicId(),
+ Collections.singletonList(partitionStateData.get())
+ );
+ this.result.complete(new ReadShareGroupStateSummaryResponse(new ReadShareGroupStateSummaryResponseData()
+ .setResults(Collections.singletonList(result))));
+ return;
+
+ // check retriable errors
+ case COORDINATOR_NOT_AVAILABLE:
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ case NOT_COORDINATOR:
+ log.warn("Received retriable error in read state RPC for key {}: {}", partitionKey(), error.message());
+ if (!readStateBackoff.canAttempt()) {
+ log.error("Exhausted max retries for read state RPC for key {} without success.", partitionKey());
+ readStateSummaryErrorReponse(error, new Exception("Exhausted max retries to complete read state RPC without success."));
+ return;
+ }
+ super.resetCoordinatorNode();
+ timer.add(new PersisterTimerTask(readStateBackoff.backOff(), this));
+ return;
+
+ default:
+ log.error("Unable to perform read state RPC for key {}: {}", partitionKey(), error.message());
+ readStateSummaryErrorReponse(error, null);
+ return;
+ }
+ }
+ }
+ }
+
+ // no response found specific topic partition
+ IllegalStateException exception = new IllegalStateException(
+ "Failed to read state for share partition " + partitionKey()
+ );
+ readStateSummaryErrorReponse(Errors.forException(exception), exception);
+ }
+
+ protected void readStateSummaryErrorReponse(Errors error, Exception exception) {
+ this.result.complete(new ReadShareGroupStateSummaryResponse(
+ ReadShareGroupStateSummaryResponse.toErrorResponseData(partitionKey().topicId(), partitionKey().partition(), error, "Error in find coordinator. " +
+ (exception == null ? error.message() : exception.getMessage()))));
+ }
+
+ @Override
+ protected void findCoordinatorErrorResponse(Errors error, Exception exception) {
+ this.result.complete(new ReadShareGroupStateSummaryResponse(
+ ReadShareGroupStateSummaryResponse.toErrorResponseData(partitionKey().topicId(), partitionKey().partition(), error, "Error in read state RPC. " +
Review Comment:
"summary"
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * The result of the {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsResult {
+
+ private final Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures;
+
+ public ListShareGroupOffsetsResult(Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures) {
+ this.futures = futures;
+ }
+
+ /**
+ * Return a future which yields all Map<String, Map<TopicPartition, Long> objects, if requests for all the groups succeed.
+ */
+ public KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
+ nil -> {
+ Map<String, Map<TopicPartition, Long>> offsets = new HashMap<>(futures.size());
+ futures.forEach((key, future) -> {
Review Comment:
It would be a bit easier to see what's going on if `key` was renamed to `groupId` since that's what the key is here.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -81,6 +83,14 @@ public interface ShareCoordinator {
*/
CompletableFuture<ReadShareGroupStateResponseData> readState(RequestContext context, ReadShareGroupStateRequestData request);
+ /**
+ * Handle read share state call
Review Comment:
"summary"
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsSpec.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Specification of share group offsets to list using {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)}.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsSpec {
+
+ private Collection<TopicPartition> topicPartitions;
+
+ public ListShareGroupOffsetsSpec() {
+ topicPartitions = new ArrayList<>();
+ }
+
+ /**
+ * Set the topic partitions whose offsets are to be listed for a share group.
+ */
+ ListShareGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions) {
+ this.topicPartitions = topicPartitions;
+ return this;
+ }
+
+ /**
+ * Returns the topic partitions whose offsets are to be listed for a share group.
+ */
+ Collection<TopicPartition> topicPartitions() {
+ return topicPartitions;
+ }
+}
Review Comment:
Please implement equals, hashCode and toString.
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryResponse.java:
##########
@@ -64,4 +67,46 @@ public static ReadShareGroupStateSummaryResponse parse(ByteBuffer buffer, short
new ReadShareGroupStateSummaryResponseData(new ByteBufferAccessor(buffer), version)
);
}
+
+ public static ReadShareGroupStateSummaryResponseData toErrorResponseData(Uuid topicId, int partitionId, Errors error, String errorMessage) {
+ return new ReadShareGroupStateSummaryResponseData().setResults(
+ Collections.singletonList(new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(Collections.singletonList(new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setPartition(partitionId)
+ .setErrorCode(error.code())
+ .setErrorMessage(errorMessage)))));
+ }
+
+ public static ReadShareGroupStateSummaryResponseData.PartitionResult toErrorResponsePartitionResult(int partitionId, Errors error, String errorMessage) {
+ return new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setPartition(partitionId)
+ .setErrorCode(error.code())
+ .setErrorMessage(errorMessage);
+ }
+
+ public static ReadShareGroupStateSummaryResponseData toResponseData(
+ Uuid topicId,
+ int partition,
+ long startOffset,
+ int stateEpoch
+ ) {
+ return new ReadShareGroupStateSummaryResponseData()
+ .setResults(Collections.singletonList(
+ new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(Collections.singletonList(
Review Comment:
Because we've recently dropped Java 8, you can now use `List.of` instead of `Collections.singletonList`.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -478,6 +481,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readState method in ShareCoordinatorShard expects a single key in the request. Hence, we will
+ // be looping over the keys below and constructing new ReadShareGroupStateRequestData objects to pass
Review Comment:
"Summary"
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -435,6 +438,60 @@ public ReadShareGroupStateResponseData readState(ReadShareGroupStateRequestData
return ReadShareGroupStateResponse.toResponseData(topicId, partition, offsetValue.startOffset(), offsetValue.stateEpoch(), stateBatches);
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ // records to read (with the key of snapshot type), response to caller
+ // only one key will be there in the request by design
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request, offset);
+ if (error.isPresent()) {
+ return error.get();
+ }
+
+ Uuid topicId = request.topics().get(0).topicId();
+ int partition = request.topics().get(0).partitions().get(0).partition();
+ int leaderEpoch = request.topics().get(0).partitions().get(0).leaderEpoch();
+
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partition);
+
+ if (!shareStateMap.containsKey(coordinatorKey)) {
+ return ReadShareGroupStateSummaryResponse.toResponseData(
+ topicId,
+ partition,
+ PartitionFactory.UNINITIALIZED_START_OFFSET,
+ PartitionFactory.DEFAULT_STATE_EPOCH
+ );
+ }
+
+ ShareGroupOffset offsetValue = shareStateMap.get(coordinatorKey, offset);
+
+ if (offsetValue == null) {
+ // Returning an error response as the snapshot value was not found
+ return ReadShareGroupStateSummaryResponse.toErrorResponseData(
+ topicId,
+ partition,
+ Errors.UNKNOWN_SERVER_ERROR,
+ "Data not found for topic {}, partition {} for group {}, in the in-memory state of share coordinator"
+ );
+ }
+
+ // Updating the leader map with the new leader epoch
+ leaderEpochMap.put(coordinatorKey, leaderEpoch);
Review Comment:
I would expect that the leader epoch is not known in this case. Since the call comes from the group coordinator and it is the not the partition leader, I think it's going to specify -1 as the epoch and the update of the leader epoch map should not occur.
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +733,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
+ }
+
+ @Override
+ protected AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> requestBuilder() {
+ throw new RuntimeException("Read Summary requests are batchable, hence individual requests not needed.");
+ }
+
+ @Override
+ protected boolean isResponseForRequest(ClientResponse response) {
+ return response.requestHeader().apiKey() == ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY;
+ }
+
+ @Override
+ protected void handleRequestResponse(ClientResponse response) {
+ log.debug("Read state summary response received - {}", response);
+ readStateBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateResult : combinedResponse.data().results()) {
+ if (readStateResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateBackoff.resetAttempts();
+ ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult result = ReadShareGroupStateSummaryResponse.toResponseReadStateSummaryResult(
+ partitionKey().topicId(),
+ Collections.singletonList(partitionStateData.get())
Review Comment:
List.of
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -478,6 +481,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readState method in ShareCoordinatorShard expects a single key in the request. Hence, we will
+ // be looping over the keys below and constructing new ReadShareGroupStateRequestData objects to pass
+ // onto the shard method.
+
+ request.topics().forEach(topicData -> {
+ Uuid topicId = topicData.topicId();
+ topicData.partitions().forEach(partitionData -> {
+ // Request object containing information of a single topic partition
+ ReadShareGroupStateSummaryRequestData requestForCurrentPartition = new ReadShareGroupStateSummaryRequestData()
+ .setGroupId(groupId)
+ .setTopics(Collections.singletonList(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
+ .setTopicId(topicId)
+ .setPartitions(Collections.singletonList(partitionData))));
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partitionData.partition());
+ // Scheduling a runtime read operation to read share partition state from the coordinator in memory state
+ CompletableFuture<ReadShareGroupStateSummaryResponseData> future = runtime.scheduleReadOperation(
+ "read-share-group-state-summary",
+ topicPartitionFor(coordinatorKey),
+ (coordinator, offset) -> coordinator.readStateSummary(requestForCurrentPartition, offset)
+ ).exceptionally(exception -> handleOperationException(
+ "read-share-group-state",
Review Comment:
"summary"
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +733,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
+ }
+
+ @Override
+ protected AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> requestBuilder() {
+ throw new RuntimeException("Read Summary requests are batchable, hence individual requests not needed.");
+ }
+
+ @Override
+ protected boolean isResponseForRequest(ClientResponse response) {
+ return response.requestHeader().apiKey() == ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY;
+ }
+
+ @Override
+ protected void handleRequestResponse(ClientResponse response) {
+ log.debug("Read state summary response received - {}", response);
+ readStateBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateResult : combinedResponse.data().results()) {
+ if (readStateResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateBackoff.resetAttempts();
+ ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult result = ReadShareGroupStateSummaryResponse.toResponseReadStateSummaryResult(
+ partitionKey().topicId(),
+ Collections.singletonList(partitionStateData.get())
+ );
+ this.result.complete(new ReadShareGroupStateSummaryResponse(new ReadShareGroupStateSummaryResponseData()
+ .setResults(Collections.singletonList(result))));
+ return;
+
+ // check retriable errors
+ case COORDINATOR_NOT_AVAILABLE:
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ case NOT_COORDINATOR:
+ log.warn("Received retriable error in read state RPC for key {}: {}", partitionKey(), error.message());
Review Comment:
"summary"
##########
share/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -278,9 +279,92 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) throws IllegalArgumentException {
- throw new RuntimeException("not implemented");
+ validate(request);
+ GroupTopicPartitionData<PartitionIdLeaderEpochData> gtp = request.groupTopicPartitionData();
+ String groupId = gtp.groupId();
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponse>>> futureMap = new HashMap<>();
+ List<PersisterStateManager.ReadStateSummaryHandler> handlers = new ArrayList<>();
+
+ gtp.topicsData().forEach(topicData -> {
+ topicData.partitions().forEach(partitionData -> {
+ CompletableFuture<ReadShareGroupStateSummaryResponse> future = futureMap
+ .computeIfAbsent(topicData.topicId(), k -> new HashMap<>())
+ .computeIfAbsent(partitionData.partition(), k -> new CompletableFuture<>());
+
+ handlers.add(
+ stateManager.new ReadStateSummaryHandler(
+ groupId,
+ topicData.topicId(),
+ partitionData.partition(),
+ partitionData.leaderEpoch(),
+ future,
+ null)
+ );
+ });
+ });
+
+ for (PersisterStateManager.PersisterStateManagerHandler handler : handlers) {
+ stateManager.enqueue(handler);
+ }
+
+ // Combine all futures into a single CompletableFuture<Void>
+ CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
+ handlers.stream()
+ .map(PersisterStateManager.ReadStateSummaryHandler::result)
+ .toArray(CompletableFuture[]::new));
+
+ // Transform the combined CompletableFuture<Void> into CompletableFuture<ReadShareGroupStateResult>
+ return combinedFuture.thenApply(v -> readSummaryResponsesToResult(futureMap));
}
+ /**
+ * Takes in a list of COMPLETED futures and combines the results,
+ * taking care of errors if any, into a single ReadShareGroupStateSummaryResult
+ * @param futureMap - HashMap of {topic -> {part -> future}}
+ * @return Object representing combined result of type ReadShareGroupStateSummaryResult
+ */
+ // visible for testing
+ ReadShareGroupStateSummaryResult readSummaryResponsesToResult(
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponse>>> futureMap
+ ) {
+ List<TopicData<PartitionStateErrorData>> topicsData = futureMap.keySet().stream()
+ .map(topicId -> {
Review Comment:
nit: I think your line continuation indentation is 8 here, but the source file uses 4.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1915646369
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.clients.admin.internals.CoordinatorKey;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+/**
+ * The result of the {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)} call.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsResult {
+
+ private final Map<String, KafkaFuture<Map<TopicPartition, Long>>> futures;
+
+ public ListShareGroupOffsetsResult(final Map<CoordinatorKey, KafkaFuture<Map<TopicPartition, Long>>> futures) {
+ this.futures = futures.entrySet().stream()
+ .collect(Collectors.toMap(e -> e.getKey().idValue, Map.Entry::getValue));
+ }
+
+ /**
+ * Return a future which yields all Map<String, Map<TopicPartition, Long> objects, if requests for all the groups succeed.
+ */
+ public KafkaFuture<Map<String, Map<TopicPartition, Long>>> all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
+ nil -> {
+ Map<String, Map<TopicPartition, Long>> offsets = new HashMap<>(futures.size());
+ futures.forEach((groupId, future) -> {
+ try {
+ offsets.put(groupId, future.get());
+ } catch (InterruptedException | ExecutionException e) {
+ // This should be unreachable, since the KafkaFuture#allOf already ensured
+ // that all the futures completed successfully.
+ throw new RuntimeException(e);
+ }
+ });
+ return offsets;
+ });
+ }
+
+ /**
+ * Return a future which yields a map of topic partitions to offsets for the specified group.
+ */
+ public KafkaFuture<Map<TopicPartition, Long>> partitionsToOffset(String groupId) {
+ if (!futures.containsKey(groupId)) {
+ throw new IllegalArgumentException("Group ID not found: " + groupId);
+ }
+ return futures.get(groupId);
+ }
+}
Review Comment:
new line in the end.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> coordinatorKeys) {
+ return AdminApiFuture.forKeys(coordinatorKeys(coordinatorKeys));
+ }
+
+ @Override
+ public String apiName() {
+ return "listShareGroupOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> coordinatorKeys) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(coordinatorKeys)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + coordinatorKeys +
+ " (expected among " + keys + ")");
+ }
+ }
+
+ @Override
+ public OffsetFetchRequest.Builder buildBatchedRequest(int brokerId, Set<CoordinatorKey> coordinatorKeys) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(coordinatorKeys.size());
+ coordinatorKeys.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? List.copyOf(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public ApiResult<CoordinatorKey, Map<TopicPartition, Long>> handleResponse(
+ Node coordinator,
+ Set<CoordinatorKey> coordinatorKeys,
+ AbstractResponse abstractResponse
+ ) {
+ validateKeys(coordinatorKeys);
+
+ final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
+
+ Map<CoordinatorKey, Map<TopicPartition, Long>> completed = new HashMap<>();
+ Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ List<CoordinatorKey> unmapped = new ArrayList<>();
+ for (CoordinatorKey coordinatorKey : coordinatorKeys) {
+ String group = coordinatorKey.idValue;
+ if (response.groupHasError(group)) {
+ handleGroupError(CoordinatorKey.byGroupId(group), response.groupLevelError(group), failed, unmapped);
+ } else {
+ final Map<TopicPartition, Long> groupOffsetsListing = new HashMap<>();
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = response.partitionDataMap(group);
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> partitionEntry : responseData.entrySet()) {
+ final TopicPartition topicPartition = partitionEntry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = partitionEntry.getValue();
+ final Errors error = partitionData.error;
+
+ if (error == Errors.NONE) {
+ final long offset = partitionData.offset;
+ groupOffsetsListing.put(topicPartition, offset);
+ } else {
+ log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
+ }
+ }
+ completed.put(CoordinatorKey.byGroupId(group), groupOffsetsListing);
+ }
+ }
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> coordinatorKeys) {
+ return coordinatorKeys.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+
+ private void handleGroupError(
+ CoordinatorKey groupId,
+ Errors error,
+ Map<CoordinatorKey, Throwable> failed,
+ List<CoordinatorKey> groupsToUnmap
+ ) {
+ switch (error) {
+ case GROUP_AUTHORIZATION_FAILED:
+ case GROUP_ID_NOT_FOUND:
+ log.debug("`OffsetFetch` request for group id {} failed due to error {}", groupId.idValue, error);
+ failed.put(groupId, error.exception());
+ break;
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ // If the coordinator is in the middle of loading, then we just need to retry
+ log.debug("`OffsetFetch` request for group id {} failed because the coordinator " +
+ "is still in the process of loading state. Will retry", groupId.idValue);
+ break;
+
+ case COORDINATOR_NOT_AVAILABLE:
+ case NOT_COORDINATOR:
+ // If the coordinator is unavailable or there was a coordinator change, then we unmap
+ // the key so that we retry the `FindCoordinator` request
+ log.debug("`OffsetFetch` request for group id {} returned error {}. " +
+ "Will attempt to find the coordinator again and retry", groupId.idValue, error);
+ groupsToUnmap.add(groupId);
+ break;
+
+ default:
+ log.error("`OffsetFetch` request for group id {} failed due to unexpected error {}", groupId.idValue, error);
+ failed.put(groupId, error.exception());
+ }
+ }
+}
Review Comment:
Line break for new file.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readStateSummary method in ShareCoordinatorShard expects a single key in the request. Hence, we will
+ // be looping over the keys below and constructing new ReadShareGroupStateSummaryRequestData objects to pass
+ // onto the shard method.
+
+ request.topics().forEach(topicData -> {
+ Uuid topicId = topicData.topicId();
+ topicData.partitions().forEach(partitionData -> {
+ // Request object containing information of a single topic partition
+ ReadShareGroupStateSummaryRequestData requestForCurrentPartition = new ReadShareGroupStateSummaryRequestData()
+ .setGroupId(groupId)
+ .setTopics(List.of(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
+ .setTopicId(topicId)
+ .setPartitions(Collections.singletonList(partitionData))));
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partitionData.partition());
+ // Scheduling a runtime read operation to read share partition state from the coordinator in memory state
+ CompletableFuture<ReadShareGroupStateSummaryResponseData> future = runtime.scheduleReadOperation(
+ "read-share-group-state-summary",
+ topicPartitionFor(coordinatorKey),
+ (coordinator, offset) -> coordinator.readStateSummary(requestForCurrentPartition, offset)
+ ).exceptionally(exception -> handleOperationException(
+ "read-share-group-state-summary",
+ request,
+ exception,
+ (error, message) -> ReadShareGroupStateSummaryResponse.toErrorResponseData(
+ topicData.topicId(),
+ partitionData.partition(),
+ error,
+ "Unable to read share group state: " + exception.getMessage()
+ ),
+ log
+ ));
+ futureMap.computeIfAbsent(topicId, k -> new HashMap<>())
+ .put(partitionData.partition(), future);
+ });
+ });
+
+ // Combine all futures into a single CompletableFuture<Void>
+ CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(futureMap.values().stream()
+ .flatMap(map -> map.values().stream()).toArray(CompletableFuture[]::new));
+
+ // Transform the combined CompletableFuture<Void> into CompletableFuture<ReadShareGroupStateResponseData>
+ return combinedFuture.thenApply(v -> {
+ List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult> readStateSummaryResult = new ArrayList<>(futureMap.size());
+ futureMap.forEach(
+ (topicId, topicEntry) -> {
+ List<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionResults = new ArrayList<>(topicEntry.size());
+ topicEntry.forEach(
+ (partitionId, responseFut) -> {
Review Comment:
nit: `responseFut` => `responseFuture`, I am not a fan of irregular variable names for saving 3 characters. Though if `fut` is well known for `future` then it's otherwise.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsOptions.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Map;
+
+/**
+ * Options for {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)}.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsOptions extends AbstractOptions<ListShareGroupOffsetsOptions> {
+}
Review Comment:
New files generally have a blank last line. Can we please add.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
Review Comment:
nit: Can we please move this `public` method prior `private`?
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +734,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
Review Comment:
Shouldn't it be `readStateSummaryHandler` as earlier it was camel case for `list...`
```suggestion
return "readStateSummaryHandler";
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java:
##########
@@ -210,6 +212,18 @@ CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGrou
List<String> groupIds
);
+ /**
+ * List share group Offsets.
+ *
+ * @param context The coordinator request context.
+ * @param request The ReadShareGroupStateSummaryRequest data.
+ *
+ * @return A future yielding the results or an exception.
+ */
+ CompletableFuture<List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult>> listShareGroupOffsets(
Review Comment:
I meant the method take `ReadShareGroupStateSummaryRequestData` as param hence I was expecting method name something like `readShareGroupStateSummary` but the current method name is `listShareGroupOffsets`, an reason?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ // records to read (with the key of snapshot type), response to caller
+ // only one key will be there in the request by design
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request, offset);
+ if (error.isPresent()) {
+ return error.get();
+ }
+
+ Uuid topicId = request.topics().get(0).topicId();
+ int partition = request.topics().get(0).partitions().get(0).partition();
+
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partition);
+
+ if (!shareStateMap.containsKey(coordinatorKey)) {
+ return ReadShareGroupStateSummaryResponse.toResponseData(
+ topicId,
+ partition,
+ PartitionFactory.UNINITIALIZED_START_OFFSET,
+ PartitionFactory.DEFAULT_STATE_EPOCH
+ );
+ }
+
+ ShareGroupOffset offsetValue = shareStateMap.get(coordinatorKey, offset);
+
Review Comment:
nit: remove line break.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> coordinatorKeys) {
+ return AdminApiFuture.forKeys(coordinatorKeys(coordinatorKeys));
+ }
+
+ @Override
+ public String apiName() {
+ return "listShareGroupOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> coordinatorKeys) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(coordinatorKeys)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + coordinatorKeys +
+ " (expected among " + keys + ")");
+ }
+ }
+
+ @Override
+ public OffsetFetchRequest.Builder buildBatchedRequest(int brokerId, Set<CoordinatorKey> coordinatorKeys) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(coordinatorKeys.size());
+ coordinatorKeys.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? List.copyOf(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public ApiResult<CoordinatorKey, Map<TopicPartition, Long>> handleResponse(
+ Node coordinator,
+ Set<CoordinatorKey> coordinatorKeys,
+ AbstractResponse abstractResponse
+ ) {
+ validateKeys(coordinatorKeys);
+
+ final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
+
+ Map<CoordinatorKey, Map<TopicPartition, Long>> completed = new HashMap<>();
+ Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ List<CoordinatorKey> unmapped = new ArrayList<>();
+ for (CoordinatorKey coordinatorKey : coordinatorKeys) {
+ String group = coordinatorKey.idValue;
+ if (response.groupHasError(group)) {
+ handleGroupError(CoordinatorKey.byGroupId(group), response.groupLevelError(group), failed, unmapped);
+ } else {
+ final Map<TopicPartition, Long> groupOffsetsListing = new HashMap<>();
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = response.partitionDataMap(group);
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> partitionEntry : responseData.entrySet()) {
+ final TopicPartition topicPartition = partitionEntry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = partitionEntry.getValue();
+ final Errors error = partitionData.error;
+
+ if (error == Errors.NONE) {
+ final long offset = partitionData.offset;
+ groupOffsetsListing.put(topicPartition, offset);
+ } else {
+ log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
+ }
+ }
+ completed.put(CoordinatorKey.byGroupId(group), groupOffsetsListing);
+ }
+ }
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> coordinatorKeys) {
+ return coordinatorKeys.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+
+ private void handleGroupError(
+ CoordinatorKey groupId,
+ Errors error,
+ Map<CoordinatorKey, Throwable> failed,
+ List<CoordinatorKey> groupsToUnmap
+ ) {
+ switch (error) {
+ case GROUP_AUTHORIZATION_FAILED:
+ case GROUP_ID_NOT_FOUND:
+ log.debug("`OffsetFetch` request for group id {} failed due to error {}", groupId.idValue, error);
+ failed.put(groupId, error.exception());
+ break;
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ // If the coordinator is in the middle of loading, then we just need to retry
+ log.debug("`OffsetFetch` request for group id {} failed because the coordinator " +
+ "is still in the process of loading state. Will retry", groupId.idValue);
+ break;
+
Review Comment:
Above case do not have an empty line after `break`, can we please have consistency.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> coordinatorKeys) {
+ return AdminApiFuture.forKeys(coordinatorKeys(coordinatorKeys));
+ }
+
+ @Override
+ public String apiName() {
+ return "listShareGroupOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> coordinatorKeys) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(coordinatorKeys)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + coordinatorKeys +
+ " (expected among " + keys + ")");
+ }
+ }
+
+ @Override
+ public OffsetFetchRequest.Builder buildBatchedRequest(int brokerId, Set<CoordinatorKey> coordinatorKeys) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(coordinatorKeys.size());
+ coordinatorKeys.forEach(g -> {
Review Comment:
instead of `g` can't we have `coordinatorKey` or `key` as the name? It's hard to read such variable names.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> coordinatorKeys) {
+ return AdminApiFuture.forKeys(coordinatorKeys(coordinatorKeys));
+ }
+
+ @Override
+ public String apiName() {
+ return "listShareGroupOffsets";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> coordinatorKeys) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(coordinatorKeys)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + coordinatorKeys +
+ " (expected among " + keys + ")");
+ }
+ }
+
+ @Override
+ public OffsetFetchRequest.Builder buildBatchedRequest(int brokerId, Set<CoordinatorKey> coordinatorKeys) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(coordinatorKeys.size());
+ coordinatorKeys.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? List.copyOf(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public ApiResult<CoordinatorKey, Map<TopicPartition, Long>> handleResponse(
+ Node coordinator,
+ Set<CoordinatorKey> coordinatorKeys,
+ AbstractResponse abstractResponse
+ ) {
+ validateKeys(coordinatorKeys);
+
+ final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
+
+ Map<CoordinatorKey, Map<TopicPartition, Long>> completed = new HashMap<>();
+ Map<CoordinatorKey, Throwable> failed = new HashMap<>();
+ List<CoordinatorKey> unmapped = new ArrayList<>();
+ for (CoordinatorKey coordinatorKey : coordinatorKeys) {
+ String group = coordinatorKey.idValue;
+ if (response.groupHasError(group)) {
+ handleGroupError(CoordinatorKey.byGroupId(group), response.groupLevelError(group), failed, unmapped);
+ } else {
+ final Map<TopicPartition, Long> groupOffsetsListing = new HashMap<>();
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData = response.partitionDataMap(group);
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> partitionEntry : responseData.entrySet()) {
+ final TopicPartition topicPartition = partitionEntry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = partitionEntry.getValue();
+ final Errors error = partitionData.error;
+
+ if (error == Errors.NONE) {
+ final long offset = partitionData.offset;
+ groupOffsetsListing.put(topicPartition, offset);
+ } else {
+ log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
+ }
+ }
+ completed.put(CoordinatorKey.byGroupId(group), groupOffsetsListing);
+ }
+ }
+ return new ApiResult<>(completed, failed, unmapped);
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> coordinatorKeys) {
+ return coordinatorKeys.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
Review Comment:
Remove additional line break
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler extends AdminApiHandler.Batched<CoordinatorKey, Map<TopicPartition, Long>> {
Review Comment:
Why don't we have unit tests for the class? Am I missing something?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java:
##########
@@ -76,6 +78,14 @@ public interface ShareCoordinator {
*/
CompletableFuture<ReadShareGroupStateResponseData> readState(RequestContext context, ReadShareGroupStateRequestData request);
+ /**
+ * Handle read share state summary call
+ * @param context - represents the incoming read request context
+ * @param request - actual RPC request object
+ * @return completable future comprising read RPC response data
Review Comment:
nit: may be it's too geeneric and you want to correct it.
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -4318,9 +4318,22 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleReadShareGroupStateSummaryRequest(request: RequestChannel.Request): Unit = {
val readShareGroupStateSummaryRequest = request.body[ReadShareGroupStateSummaryRequest]
- // TODO: Implement the ReadShareGroupStateSummaryRequest handling
- requestHelper.sendMaybeThrottle(request, readShareGroupStateSummaryRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
- CompletableFuture.completedFuture[Unit](())
+ authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
+
+ shareCoordinator match {
+ case None => requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+ readShareGroupStateSummaryRequest.getErrorResponse(requestThrottleMs,
+ new ApiException("Share coordinator is not enabled.")))
+ CompletableFuture.completedFuture[Unit](())
+ case Some(coordinator) => coordinator.readStateSummary(request.context, readShareGroupStateSummaryRequest.data)
+ .handle[Unit] { (response, exception) =>
+ if (exception != null) {
+ requestHelper.sendMaybeThrottle(request, readShareGroupStateSummaryRequest.getErrorResponse(exception))
+ } else {
+ requestHelper.sendMaybeThrottle(request, new ReadShareGroupStateSummaryResponse(response))
+ }
+ }
+ }
Review Comment:
I can't see the KafkaApisTest for this change, am I missing something?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
Review Comment:
Why this check is not prior to topic list iteration? If this is true then why to iterate for partitions.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
Review Comment:
Same as above.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
Review Comment:
The log is an error and response is empty (without exception), is it intended?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(
+ generateErrorReadStateSummaryResponse(
+ request,
+ Errors.COORDINATOR_NOT_AVAILABLE,
+ "Share coordinator is not available."
+ )
+ );
+ }
+
+ // The request received here could have multiple keys of structure group:topic:partition. However,
+ // the readStateSummary method in ShareCoordinatorShard expects a single key in the request. Hence, we will
+ // be looping over the keys below and constructing new ReadShareGroupStateSummaryRequestData objects to pass
+ // onto the shard method.
+
+ request.topics().forEach(topicData -> {
+ Uuid topicId = topicData.topicId();
+ topicData.partitions().forEach(partitionData -> {
+ // Request object containing information of a single topic partition
+ ReadShareGroupStateSummaryRequestData requestForCurrentPartition = new ReadShareGroupStateSummaryRequestData()
+ .setGroupId(groupId)
+ .setTopics(List.of(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
+ .setTopicId(topicId)
+ .setPartitions(Collections.singletonList(partitionData))));
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partitionData.partition());
+ // Scheduling a runtime read operation to read share partition state from the coordinator in memory state
+ CompletableFuture<ReadShareGroupStateSummaryResponseData> future = runtime.scheduleReadOperation(
+ "read-share-group-state-summary",
+ topicPartitionFor(coordinatorKey),
+ (coordinator, offset) -> coordinator.readStateSummary(requestForCurrentPartition, offset)
+ ).exceptionally(exception -> handleOperationException(
+ "read-share-group-state-summary",
+ request,
+ exception,
+ (error, message) -> ReadShareGroupStateSummaryResponse.toErrorResponseData(
+ topicData.topicId(),
+ partitionData.partition(),
+ error,
+ "Unable to read share group state: " + exception.getMessage()
+ ),
+ log
+ ));
+ futureMap.computeIfAbsent(topicId, k -> new HashMap<>())
+ .put(partitionData.partition(), future);
+ });
+ });
+
+ // Combine all futures into a single CompletableFuture<Void>
+ CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(futureMap.values().stream()
+ .flatMap(map -> map.values().stream()).toArray(CompletableFuture[]::new));
+
+ // Transform the combined CompletableFuture<Void> into CompletableFuture<ReadShareGroupStateResponseData>
+ return combinedFuture.thenApply(v -> {
+ List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult> readStateSummaryResult = new ArrayList<>(futureMap.size());
+ futureMap.forEach(
+ (topicId, topicEntry) -> {
+ List<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionResults = new ArrayList<>(topicEntry.size());
+ topicEntry.forEach(
+ (partitionId, responseFut) -> {
+ // responseFut would already be completed by now since we have used
+ // CompletableFuture::allOf to create a combined future from the future map.
+ partitionResults.add(
+ responseFut.getNow(null).results().get(0).partitions().get(0)
Review Comment:
nit: you might want to correct the indentation.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -712,6 +737,54 @@ public CompletableFuture<List<DescribedGroup>> shareGroupDescribe(
return FutureUtils.combineFutures(futures, ArrayList::new, List::addAll);
}
+ /**
+ * See {@link GroupCoordinator#listShareGroupOffsets(RequestContext, ReadShareGroupStateSummaryRequestData)}.
+ */
+ @Override
+ public CompletableFuture<List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult>> listShareGroupOffsets(
+ RequestContext context,
+ ReadShareGroupStateSummaryRequestData requestData
+ ) {
Review Comment:
Do we have unit tests for the method? I can't find any though.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
Review Comment:
Shouldn't this be created when used i.e. atleast after the `isEmpty` check. If code returns from `if (isEmpty(request.topics())) {` then why to initialize the map.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,113 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ String groupId = request.groupId();
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // Send an empty response if topic data is empty
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if partition data is empty for any topic
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ if (isEmpty(topicData.partitions())) {
+ log.error("Partition Data for topic {} is empty: {}", topicData.topicId(), request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+ }
+
+ // Send an empty response if groupId is invalid
+ if (isGroupIdEmpty(groupId)) {
+ log.error("Group id must be specified and non-empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
+ );
+ }
+
+ // Send an empty response if the coordinator is not active
+ if (!isActive.get()) {
Review Comment:
This should be the first check in the method, even prior looking at request.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
Review Comment:
Is it correct?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ // records to read (with the key of snapshot type), response to caller
+ // only one key will be there in the request by design
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request, offset);
+ if (error.isPresent()) {
+ return error.get();
+ }
+
+ Uuid topicId = request.topics().get(0).topicId();
+ int partition = request.topics().get(0).partitions().get(0).partition();
Review Comment:
Similar as above.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ // records to read (with the key of snapshot type), response to caller
+ // only one key will be there in the request by design
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request, offset);
+ if (error.isPresent()) {
+ return error.get();
+ }
+
+ Uuid topicId = request.topics().get(0).topicId();
Review Comment:
Is it guaranteed that the request will have the topic and that too only 1? Or do we need to validate?
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +734,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
Review Comment:
Can we please have javadoc for class.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -573,6 +626,46 @@ private Optional<ReadShareGroupStateResponseData> maybeGetReadStateError(ReadSha
return Optional.empty();
}
+ private Optional<ReadShareGroupStateSummaryResponseData> maybeGetReadStateSummaryError(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ String groupId = request.groupId();
+ ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData = request.topics().get(0);
+ ReadShareGroupStateSummaryRequestData.PartitionData partitionData = topicData.partitions().get(0);
+
+ Uuid topicId = topicData.topicId();
+ int partitionId = partitionData.partition();
+
+ if (topicId == null) {
+ log.error("Request topic id is null.");
Review Comment:
Are these logs helpful and should be logged as `error`?
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -573,6 +626,46 @@ private Optional<ReadShareGroupStateResponseData> maybeGetReadStateError(ReadSha
return Optional.empty();
}
+ private Optional<ReadShareGroupStateSummaryResponseData> maybeGetReadStateSummaryError(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ String groupId = request.groupId();
+ ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData = request.topics().get(0);
+ ReadShareGroupStateSummaryRequestData.PartitionData partitionData = topicData.partitions().get(0);
Review Comment:
Same as above here.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -493,6 +496,56 @@ private List<PersisterStateBatch> mergeBatches(
.combineStateBatches();
}
+ /**
+ * This method finds the ShareSnapshotValue record corresponding to the requested topic partition from the
+ * in-memory state of coordinator shard, the shareStateMap.
+ * <p>
+ * This method as called by the ShareCoordinatorService will be provided with
+ * the request data which covers only key i.e. group1:topic1:partition1. The implementation
+ * below was done keeping this in mind.
+ *
+ * @param request - ReadShareGroupStateSummaryRequestData for a single key
+ * @param offset - offset to read from the __share_group_state topic partition
+ * @return CoordinatorResult(records, response)
+ */
+ public ReadShareGroupStateSummaryResponseData readStateSummary(ReadShareGroupStateSummaryRequestData request, Long offset) {
+ // records to read (with the key of snapshot type), response to caller
+ // only one key will be there in the request by design
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request, offset);
+ if (error.isPresent()) {
+ return error.get();
+ }
+
+ Uuid topicId = request.topics().get(0).topicId();
+ int partition = request.topics().get(0).partitions().get(0).partition();
+
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partition);
+
+ if (!shareStateMap.containsKey(coordinatorKey)) {
+ return ReadShareGroupStateSummaryResponse.toResponseData(
+ topicId,
+ partition,
+ PartitionFactory.UNINITIALIZED_START_OFFSET,
+ PartitionFactory.DEFAULT_STATE_EPOCH
+ );
+ }
+
+ ShareGroupOffset offsetValue = shareStateMap.get(coordinatorKey, offset);
+
+ if (offsetValue == null) {
+ // Returning an error response as the snapshot value was not found
+ return ReadShareGroupStateSummaryResponse.toErrorResponseData(
+ topicId,
+ partition,
+ Errors.UNKNOWN_SERVER_ERROR,
Review Comment:
Is it the right error or we need to be specific about missing key? Also shouldn't we log in such cases as `unknown server error` is returned which means that this scenario should not occur in broker.
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +734,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
Review Comment:
Is the indentation correct?
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +734,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
+ }
+
+ @Override
+ protected AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> requestBuilder() {
+ throw new RuntimeException("Read Summary requests are batchable, hence individual requests not needed.");
Review Comment:
Should it be RuntimeException or IllegalArgumentException, though depends what the final response from caller would be?
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -730,6 +734,147 @@ protected RPCType rpcType() {
}
}
+ public class ReadStateSummaryHandler extends PersisterStateManagerHandler {
+ private final int leaderEpoch;
+ private final CompletableFuture<ReadShareGroupStateSummaryResponse> result;
+ private final BackoffManager readStateBackoff;
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ long backoffMs,
+ long backoffMaxMs,
+ int maxRPCRetryAttempts,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ super(groupId, topicId, partition, backoffMs, backoffMaxMs, maxRPCRetryAttempts);
+ this.leaderEpoch = leaderEpoch;
+ this.result = result;
+ this.readStateBackoff = new BackoffManager(maxRPCRetryAttempts, backoffMs, backoffMaxMs);
+ }
+
+ public ReadStateSummaryHandler(
+ String groupId,
+ Uuid topicId,
+ int partition,
+ int leaderEpoch,
+ CompletableFuture<ReadShareGroupStateSummaryResponse> result,
+ Consumer<ClientResponse> onCompleteCallback
+ ) {
+ this(
+ groupId,
+ topicId,
+ partition,
+ leaderEpoch,
+ result,
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_FIND_COORD_ATTEMPTS,
+ onCompleteCallback
+ );
+ }
+
+ @Override
+ protected String name() {
+ return "ReadStateSummaryHandler";
+ }
+
+ @Override
+ protected AbstractRequest.Builder<ReadShareGroupStateSummaryRequest> requestBuilder() {
+ throw new RuntimeException("Read Summary requests are batchable, hence individual requests not needed.");
+ }
+
+ @Override
+ protected boolean isResponseForRequest(ClientResponse response) {
+ return response.requestHeader().apiKey() == ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY;
+ }
+
+ @Override
+ protected void handleRequestResponse(ClientResponse response) {
Review Comment:
Do we have tests for these changes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1904176664
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsSpec.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Specification of share group offsets to list using {@link Admin#listShareGroupOffsets(Map, ListShareGroupOffsetsOptions)}.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListShareGroupOffsetsSpec {
+
+ private Collection<TopicPartition> topicPartitions;
+
+ public ListShareGroupOffsetsSpec() {
+ topicPartitions = new ArrayList<>();
+ }
+
+ /**
+ * Set the topic partitions whose offsets are to be listed for a share group.
+ */
+ ListShareGroupOffsetsSpec topicPartitions(Collection<TopicPartition> topicPartitions) {
+ this.topicPartitions = topicPartitions;
+ return this;
+ }
+
+ /**
+ * Returns the topic partitions whose offsets are to be listed for a share group.
+ */
+ public Collection<TopicPartition> topicPartitions() {
+ return topicPartitions;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ListShareGroupOffsetsSpec)) {
+ return false;
+ }
+ ListShareGroupOffsetsSpec that = (ListShareGroupOffsetsSpec) o;
+ return Objects.equals(topicPartitions, that.topicPartitions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicPartitions);
+ }
+
+ @Override
+ public String toString() {
+ return "ListShareGroupOffsetsSpec(" +
+ "topicPartitions=" + topicPartitions +
Review Comment:
Will a null `topicPartitions` generate an error? I can see an override where `topicPartitions` can be set to null overriding the constructor initialization.
```
topicPartitions(Collection<TopicPartition> topicPartitions) {
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3796,6 +3797,17 @@ public DescribeShareGroupsResult describeShareGroups(final Collection<String> gr
.collect(Collectors.toMap(entry -> entry.getKey().idValue, Map.Entry::getValue)));
}
+ @Override
+ public ListShareGroupOffsetsResult listShareGroupOffsets(Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ ListShareGroupOffsetsOptions options) {
+ SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> future =
+ ListShareGroupOffsetsHandler.newFuture(groupSpecs.keySet());
+ ListShareGroupOffsetsHandler handler =
+ new ListShareGroupOffsetsHandler(groupSpecs, logContext);
Review Comment:
nit: merge lines and maybe the above ones as well.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
Review Comment:
Isn't `listShareGroupOffsets` be a better name?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
Review Comment:
For my undersatnding: ListOffsetsHandled extends Batched which implements AdminApiHandler, hence is ListShareGroupOffsetsHandler differ in functionality with list offsets?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
Review Comment:
nit: Can Public methods be prior to protected and private? i.e. write public > protected > private, it's just easier that way to look at code files.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
Review Comment:
`(expected one of ` => `(expected among` as it's a Set not just one?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(groupIds.size());
+ groupIds.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? new ArrayList<>(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, false, false);
+ }
+
+ @Override
+ public Collection<RequestAndKeys<CoordinatorKey>> buildRequest(int brokerId, Set<CoordinatorKey> groupIds) {
Review Comment:
Set<CoordinatorKey> groupIds => Set<CoordinatorKey> coordinatorKeys? Elsewhere as well, it's just confusing to see groupIds.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
Review Comment:
`Set<CoordinatorKey> groupIds` => `Set<CoordinatorKey> coordinatorKeys`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -175,6 +189,8 @@ public GroupCoordinatorService build() {
throw new IllegalArgumentException("GroupCoordinatorMetrics must be set.");
if (groupConfigManager == null)
throw new IllegalArgumentException("GroupConfigManager must be set.");
+ if (persister == null)
+ throw new IllegalArgumentException("Persister must be set.");
Review Comment:
Does that mean a Persister is always needed regardless share group is enabled on the broker?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java:
##########
@@ -210,6 +212,18 @@ CompletableFuture<List<ShareGroupDescribeResponseData.DescribedGroup>> shareGrou
List<String> groupIds
);
+ /**
+ * List share group Offsets.
+ *
+ * @param context The coordinator request context.
+ * @param request The ReadShareGroupStateSummaryRequest data.
+ *
+ * @return A future yielding the results or an exception.
+ */
+ CompletableFuture<List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult>> listShareGroupOffsets(
Review Comment:
Shouldn't the method align more with `ReadShareGroupStateSummaryRequest` operation name rather with list group offsets i.e. `readShareGroupStateSummary`?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListShareGroupOffsetsHandler.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.OffsetFetchRequest;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListShareGroupOffsetsHandler implements AdminApiHandler<CoordinatorKey, Map<TopicPartition, Long>> {
+
+ private final Map<String, ListShareGroupOffsetsSpec> groupSpecs;
+ private final Logger log;
+ private final CoordinatorStrategy lookupStrategy;
+
+ public ListShareGroupOffsetsHandler(
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(ListShareGroupOffsetsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, Map<TopicPartition, Long>> newFuture(Collection<String> groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
+ }
+
+ @Override
+ public String apiName() {
+ return "offsetFetch";
+ }
+
+ @Override
+ public AdminApiLookupStrategy<CoordinatorKey> lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ private void validateKeys(Set<CoordinatorKey> groupIds) {
+ Set<CoordinatorKey> keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
+ throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
+ " (expected one of " + keys + ")");
+ }
+ }
+
+ private static Set<CoordinatorKey> coordinatorKeys(Collection<String> groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set<CoordinatorKey> groupIds) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map<String, List<TopicPartition>> coordinatorGroupIdToTopicPartitions = new HashMap<>(groupIds.size());
+ groupIds.forEach(g -> {
+ ListShareGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List<TopicPartition> partitions = spec.topicPartitions() != null ? new ArrayList<>(spec.topicPartitions()) : null;
Review Comment:
Should the list be `mutable`? If not then shall we use `List.copyOf(spec.topicPartitions())`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DRAFT] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #17775:
URL: https://github.com/apache/kafka/pull/17775#issuecomment-2475717791
cc: @AndrewJSchofield for an initial review on the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DRAFT] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1863265749
##########
clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java:
##########
@@ -304,6 +304,12 @@ public DescribeShareGroupsResult describeShareGroups(Collection<String> groupIds
return delegate.describeShareGroups(groupIds, options);
}
+ @Override
+ public ListShareGroupOffsetsResult listShareGroupOffsets(Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ ListShareGroupOffsetsOptions options) {
Review Comment:
nit: Slightly scruffy indentation compared with the rest of this file.
##########
checkstyle/checkstyle.xml:
##########
@@ -123,7 +123,7 @@
</module>
<module name="ClassDataAbstractionCoupling">
<!-- default is 7 -->
- <property name="max" value="25"/>
+ <property name="max" value="30"/>
Review Comment:
This seems like a surprising side-effect of this PR. I don't think you should change this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA 16720 : AdminClient support for ListShareGroupOffsets [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #17775:
URL: https://github.com/apache/kafka/pull/17775#discussion_r1908541732
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -244,6 +261,11 @@ public GroupCoordinatorService build() {
*/
private final GroupConfigManager groupConfigManager;
+ /**
+ * The Persister to persist the state of GC
+ */
+ private final Persister persister;
Review Comment:
The persister here is needed to make the call to the persister.readSummary() method in the `listShareGroupOffsets()` function.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org