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 2025/01/16 05:37:17 UTC
[PR] [KAFKA-16720] AdminClient Support for ListShareGroupOffsets (1/n) [kafka]
sjhajharia opened a new pull request, #18571:
URL: https://github.com/apache/kafka/pull/18571
KAFKA-16720 aims at adding the support for the ListShareGroupOffsets AdminClient functionality. We started off with https://github.com/apache/kafka/pull/17775 but the same was getting a bit out of hand due to large number of file changes. Hence, to ensure smoother review, I am splitting it into two parts (hopefully)
The Flow as per the KIP:
This PR mainly focusses on the handling of the ReadShareGroupStateSummary RPC which the Persister is supposed to issue to the Share Coordinator.
Key Changes in this PR:
- Introduction of the 3 files related to ListShareGroupOffsets (`ListShareGroupOffsetsOptions`, `ListShareGroupOffsetsResult`, `ListShareGroupOffsetsSpec`).
- Introduction of the `Admin.java` function `listShareGroupOffsets()` and the corresponding changes to all inherited files.
- Addition of some utility functions to `ReadShareGroupStateSummaryRequest` and `ReadShareGroupStateSummaryResponse`.
- Handling of the `ReadShareGroupStateSummaryRequest` RPC in the `KafkaApis` (and the corresponding changes in `KafkaApisTest`).
- Changes in the ShareCoordinator classes for the handling of this RPC. (This includes `ShareCoordinator`, `ShareCoordinatorShard` and `ShareCoordinatorService`). Unit tests were added to each of their respective test files.
- Introduction of the `ReadStateSummaryHandler` in `PersisterStateManager`. Also changes to the `PersisterStateManagerTest` for the same.
- Implementation of the `readSummary()` method in `DefaultStatePersister` and the changes in the Test file.
--
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 (1/n) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1920266608
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -912,6 +1057,8 @@ public static AbstractRequest.Builder<? extends AbstractRequest> coalesceRequest
return coalesceWrites(groupId, handlers);
case READ:
return coalesceReads(groupId, handlers);
+ case SUMMARY:
+ return coalesceReadSummarys(groupId, handlers);
Review Comment:
nit: Summaries
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -577,6 +580,116 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ // 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."
+ )
+ );
+ }
+
+ String groupId = request.groupId();
+ // 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 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()
+ );
+ }
+ }
+
+ // A map to store the futures for each topicId and partition.
+ Map<Uuid, Map<Integer, CompletableFuture<ReadShareGroupStateSummaryResponseData>>> futureMap = new HashMap<>();
+
+ // 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.
+
+ for (ReadShareGroupStateSummaryRequestData.ReadStateSummaryData topicData : request.topics()) {
+ Uuid topicId = topicData.topicId();
+ for (ReadShareGroupStateSummaryRequestData.PartitionData partitionData : topicData.partitions()) {
+ SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partitionData.partition());
+
+ ReadShareGroupStateSummaryRequestData requestForCurrentPartition = new ReadShareGroupStateSummaryRequestData()
+ .setGroupId(groupId)
+ .setTopics(Collections.singletonList(new ReadShareGroupStateSummaryRequestData.ReadStateSummaryData()
Review Comment:
Since we now have Java 17 on the broker, we prefer `List.of` instead of `Collections.singletonList`.
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -398,9 +402,70 @@ public CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord> rea
return new CoordinatorResult<>(Collections.singletonList(record), responseData);
}
+ /**
+ * 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
+ * @return CoordinatorResult(records, response)
+ */
+
+ public CoordinatorResult<ReadShareGroupStateSummaryResponseData, CoordinatorRecord> readStateSummary(
+ ReadShareGroupStateSummaryRequestData request
+ ) {
+ // Only one key will be there in the request by design.
+ Optional<ReadShareGroupStateSummaryResponseData> error = maybeGetReadStateSummaryError(request);
+ if (error.isPresent()) {
+ return new CoordinatorResult<>(Collections.emptyList(), error.get());
Review Comment:
We prefer `List.of()` these days.
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,97 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
+ return CompletableFuture.failedFuture(e);
+ }
+
+ 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 -> {partition -> 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 -> {
+ List<PartitionStateErrorData> partitionStateErrorData = futureMap.get(topicId).entrySet().stream()
+ .map(partitionFuture -> {
+ int partition = partitionFuture.getKey();
+ CompletableFuture<ReadShareGroupStateSummaryResponse> future = partitionFuture.getValue();
+ try {
+ // already completed because of allOf call in the caller
+ ReadShareGroupStateSummaryResponse partitionResponse = future.join();
+ return partitionResponse.data().results().get(0).partitions().stream()
+ .map(partitionResult -> PartitionFactory.newPartitionStateErrorData(
Review Comment:
I think you need to create a new method in `PartitionFactory` that takes partition, state epoch, start offset and leader epoch as parameters. This is a success case that is subtly different than what went before.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsResult.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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) {
Review Comment:
This constructor should be package-private.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918582979
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -34,7 +35,7 @@ public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateS
private final ReadShareGroupStateSummaryRequestData data;
public Builder(ReadShareGroupStateSummaryRequestData data) {
- this(data, false);
+ this(data, true);
Review Comment:
Yes, this change seems almost inevitable. The `PersisterStateManager` class has the freshly introduced `coalesceReadSummarys()` method which generated the `ReadShareGroupStateSummaryRequest`. If we keep it as false, this request will not reach the broker as the latest version is unstable.
For the [ReadShareGroupState](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateRequest.java#L36-L38) and [WriteShareGroupState](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/WriteShareGroupStateRequest.java#L36-L39) too, it has been set to true.
The version can be passed in the `.build()` method as is the case with ShareFetchRequest and ShareAcknowledge, however, for both of these, the tests lie in the KafkaApisTest where as you mentioned, they override the property using `ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG -> "true"`.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1920361167
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,97 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
+ return CompletableFuture.failedFuture(e);
+ }
+
+ 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 -> {partition -> 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 -> {
+ List<PartitionStateErrorData> partitionStateErrorData = futureMap.get(topicId).entrySet().stream()
+ .map(partitionFuture -> {
+ int partition = partitionFuture.getKey();
+ CompletableFuture<ReadShareGroupStateSummaryResponse> future = partitionFuture.getValue();
+ try {
+ // already completed because of allOf call in the caller
+ ReadShareGroupStateSummaryResponse partitionResponse = future.join();
+ return partitionResponse.data().results().get(0).partitions().stream()
+ .map(partitionResult -> PartitionFactory.newPartitionStateErrorData(
Review Comment:
I have added a new method for the same called `newPartitionStateSummaryData`. In addition I have added a class `PartitionStateSummaryData` which contains this data. Correspondingly, the `ReadShareGroupStateSummaryResult` now uses this class instead of the old `PartitionStateErrorData`
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1919614030
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -34,7 +35,7 @@ public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateS
private final ReadShareGroupStateSummaryRequestData data;
public Builder(ReadShareGroupStateSummaryRequestData data) {
- this(data, false);
+ this(data, true);
Review Comment:
Well I agree. I have updated the code in the PersisterStateManager to pass `true` rather than update here.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1919613483
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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);
+ readStateSummaryBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateSummaryResult : combinedResponse.data().results()) {
+ if (readStateSummaryResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateSummaryResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateSummaryBackoff.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 summary RPC for key {}: {}", partitionKey(), error.message());
+ if (!readStateSummaryBackoff.canAttempt()) {
+ log.error("Exhausted max retries for read state summary RPC for key {} without success.", partitionKey());
+ readStateSummaryErrorReponse(error, new Exception("Exhausted max retries to complete read state summary RPC without success."));
+ return;
+ }
+ super.resetCoordinatorNode();
+ timer.add(new PersisterTimerTask(readStateSummaryBackoff.backOff(), this));
+ return;
+
+ default:
+ log.error("Unable to perform read state summary RPC for key {}: {}", partitionKey(), error.message());
+ readStateSummaryErrorReponse(error, null);
+ return;
+ }
+ }
+ }
+ }
+
+ // no response found specific topic partition
+ IllegalStateException exception = new IllegalStateException(
+ "Failed to read state summary for share partition " + partitionKey()
+ );
+ readStateSummaryErrorReponse(Errors.forException(exception), exception);
Review Comment:
We already have a `readStateSummaryErrorReponse()` method which accepts two parameters. I kept it like this to prevent the addition of another overloaded fucntion.
--
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 (1/n) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #18571:
URL: https://github.com/apache/kafka/pull/18571
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918494836
##########
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:
It does have an empty line after this. Else it would be followed by a red minus as is [here](https://github.com/apache/kafka/pull/17775/files#diff-27c3c80932bea29a77cefb0e32be13b9f2f398bb10f3a107da1b50061129f5e6).
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918514827
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -83,8 +84,24 @@ public ReadShareGroupStateSummaryRequestData data() {
public static ReadShareGroupStateSummaryRequest parse(ByteBuffer buffer, short version) {
return new ReadShareGroupStateSummaryRequest(
- new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
- version
+ new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
+ version
);
}
+
+ public static List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult> getErrorList(
+ List<Uuid> topicIds,
+ Errors error
+ ) {
+ return topicIds.stream()
+ .map(topicId ->
+ new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(
+ new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setErrorCode(error.code())
+ .setErrorMessage(error.exceptionName())
Review Comment:
Yes, I agree, the actual error would make more sense. Making the change.
--
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 (1/n) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918186242
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3796,6 +3796,13 @@ public DescribeShareGroupsResult describeShareGroups(final Collection<String> gr
.collect(Collectors.toMap(entry -> entry.getKey().idValue, Map.Entry::getValue)));
}
+ // To do in a follow-up PR
+ @Override
+ public ListShareGroupOffsetsResult listShareGroupOffsets(final Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ final ListShareGroupOffsetsOptions options) {
+ // To-do
+ return null;
Review Comment:
nit: generally better to throw Exception that not yet implemented but it's fine for now as follow up PRs will be soon after.
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -34,7 +35,7 @@ public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateS
private final ReadShareGroupStateSummaryRequestData data;
public Builder(ReadShareGroupStateSummaryRequestData data) {
- this(data, false);
+ this(data, true);
Review Comment:
Should this be `true`? I see ShareFetchRequest or ShareAcknowledgeRequest still has `false` in enable unstable version and get's tackled at caller level with enable versions in server.properties.
##########
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.
Review Comment:
What should one look for in Admin.java to get details about `evolving`?
##########
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:
Empty line at end of new files.
##########
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<>();
+ }
Review Comment:
nit: What's the benefit of initializing by default? I don't see any `add` method to put topicPartition in collection rather a method to completely replace.
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,96 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
+ return CompletableFuture.failedFuture(e);
+ }
+ 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}}
Review Comment:
Please do not use such abbreviations in javadoc.
```suggestion
* @param futureMap - HashMap of {topic -> {partition -> future}}
```
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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:
Though not for this PR but just `handleResponse` method name seems better.
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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 something more specific?
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,96 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
Review Comment:
Should we have error log for this failure, is this error concerning?
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -83,8 +84,24 @@ public ReadShareGroupStateSummaryRequestData data() {
public static ReadShareGroupStateSummaryRequest parse(ByteBuffer buffer, short version) {
return new ReadShareGroupStateSummaryRequest(
- new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
- version
+ new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
+ version
);
}
+
+ public static List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult> getErrorList(
+ List<Uuid> topicIds,
+ Errors error
+ ) {
+ return topicIds.stream()
+ .map(topicId ->
+ new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(
+ new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setErrorCode(error.code())
+ .setErrorMessage(error.exceptionName())
Review Comment:
Why do we want to return the class name of the exception? Also shouldn't the actual exception be passed here to get the correct error message?
##########
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.
+ */
+ public 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;
Review Comment:
In case you remove the default initialization then `return topicPartitions == null ? Collections.emptyList()/List.of() : topicPartitions;`
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,96 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
+ return CompletableFuture.failedFuture(e);
+ }
+ GroupTopicPartitionData<PartitionIdLeaderEpochData> gtp = request.groupTopicPartitionData();
Review Comment:
nit: line break above would be good.
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryResponse.java:
##########
@@ -59,9 +61,52 @@ public int throttleTimeMs() {
public void maybeSetThrottleTimeMs(int throttleTimeMs) {
// No op
}
+
public static ReadShareGroupStateSummaryResponse parse(ByteBuffer buffer, short version) {
return new ReadShareGroupStateSummaryResponse(
- new ReadShareGroupStateSummaryResponseData(new ByteBufferAccessor(buffer), version)
+ new ReadShareGroupStateSummaryResponseData(new ByteBufferAccessor(buffer), version)
);
}
+
+ public static ReadShareGroupStateSummaryResponseData toErrorResponseData(Uuid topicId, int partitionId, Errors error, String errorMessage) {
+ return new ReadShareGroupStateSummaryResponseData().setResults(
+ List.of(new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(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(List.of(
+ new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(
+ new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setPartition(partition)
+ .setStartOffset(startOffset)
+ .setStateEpoch(stateEpoch)
+ ))
+ ));
+ }
+
+ public static ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult toResponseReadStateSummaryResult(Uuid topicId, List<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionResults) {
Review Comment:
Might want to break in 2 lines, seems too big for single line.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1917779630
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java:
##########
@@ -578,6 +581,116 @@ public CompletableFuture<ReadShareGroupStateResponseData> readState(RequestConte
});
}
+ @Override
+ public CompletableFuture<ReadShareGroupStateSummaryResponseData> readStateSummary(RequestContext context, ReadShareGroupStateSummaryRequestData request) {
+ // 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."
+ )
+ );
+ }
+
+ String groupId = request.groupId();
+ // 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 topic data is empty.
+ if (isEmpty(request.topics())) {
+ log.error("Topic Data is empty: {}", request);
+ return CompletableFuture.completedFuture(
+ new ReadShareGroupStateSummaryResponseData()
Review Comment:
@apoorvmittal10 you had mentioned in the previous PR:
> The log is an error and response is empty (without exception), is it intended?
That is based on the handling which this file uses in the readState() function
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -573,6 +638,39 @@ private Optional<ReadShareGroupStateResponseData> maybeGetReadStateError(ReadSha
return Optional.empty();
}
+ private Optional<ReadShareGroupStateSummaryResponseData> maybeGetReadStateSummaryError(ReadShareGroupStateSummaryRequestData request) {
+ 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:
All these logs are mentioned as error logs based on the corresponding WriteState and ReadState methods
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1917783980
##########
share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -704,14 +708,14 @@ protected void handleRequestResponse(ClientResponse response) {
protected void readStateErrorReponse(Errors error, Exception exception) {
this.result.complete(new ReadShareGroupStateResponse(
- ReadShareGroupStateResponse.toErrorResponseData(partitionKey().topicId(), partitionKey().partition(), error, "Error in find coordinator. " +
+ ReadShareGroupStateResponse.toErrorResponseData(partitionKey().topicId(), partitionKey().partition(), error, "Error in read state RPC. " +
Review Comment:
This change was made intentionally to align the function name with the actual error msg.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918540660
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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:
We can have a separate discussion on this in that case. For the time being, I kept it as is.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918539274
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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:
This too is based on the error as seen in the [WriteStateHandler](https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java#L497-L499) and [ReadStateHandler](https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java#L641-L643).
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918531681
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,96 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
+ return CompletableFuture.failedFuture(e);
+ }
+ 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}}
Review Comment:
Ack
Will make changes to the java docs for the read and write RPCs too.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918529558
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -288,7 +291,96 @@ public CompletableFuture<DeleteShareGroupStateResult> deleteState(DeleteShareGro
* @return A completable future of ReadShareGroupStateSummaryResult
*/
public CompletableFuture<ReadShareGroupStateSummaryResult> readSummary(ReadShareGroupStateSummaryParameters request) {
- throw new RuntimeException("not implemented");
+ try {
+ validate(request);
+ } catch (Exception e) {
+ log.error("Unable to validate read state summary request", e);
Review Comment:
I am not sure about the severity here. I have tried to maintain consistency here with the [readShare()](https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java#L177-L183) and [writeShare()](https://github.com/apache/kafka/blob/trunk/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java#L81-L86) methods in the same class.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1919606461
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -969,5 +1116,27 @@ private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReads(
.setPartitions(entry.getValue()))
.collect(Collectors.toList())));
}
+
+ private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReadSummarys(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+ Map<Uuid, List<ReadShareGroupStateSummaryRequestData.PartitionData>> partitionData = new HashMap<>();
+ handlers.forEach(persHandler -> {
Review Comment:
fixed that
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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);
+ readStateSummaryBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateSummaryResult : combinedResponse.data().results()) {
+ if (readStateSummaryResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateSummaryResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateSummaryBackoff.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 summary RPC for key {}: {}", partitionKey(), error.message());
+ if (!readStateSummaryBackoff.canAttempt()) {
+ log.error("Exhausted max retries for read state summary RPC for key {} without success.", partitionKey());
+ readStateSummaryErrorReponse(error, new Exception("Exhausted max retries to complete read state summary RPC without success."));
+ return;
+ }
+ super.resetCoordinatorNode();
+ timer.add(new PersisterTimerTask(readStateSummaryBackoff.backOff(), this));
+ return;
+
+ default:
+ log.error("Unable to perform read state summary RPC for key {}: {}", partitionKey(), error.message());
+ readStateSummaryErrorReponse(error, null);
+ return;
+ }
+ }
+ }
+ }
+
+ // no response found specific topic partition
+ IllegalStateException exception = new IllegalStateException(
+ "Failed to read state summary 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 read state summary RPC. " +
+ (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 find coordinator. " +
+ (exception == null ? error.message() : exception.getMessage()))));
+ }
Review Comment:
The ReadStateSummaryHandler also makes a call to the FindCoordinator. Hence I think its good to have here.
--
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 (1/n) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1919738993
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3796,6 +3796,14 @@ public DescribeShareGroupsResult describeShareGroups(final Collection<String> gr
.collect(Collectors.toMap(entry -> entry.getKey().idValue, Map.Entry::getValue)));
}
+ // To do in a follow-up PR
+ @Override
+ public ListShareGroupOffsetsResult listShareGroupOffsets(final Map<String, ListShareGroupOffsetsSpec> groupSpecs,
+ final ListShareGroupOffsetsOptions options) {
+ // To-do
+ throw new InvalidRequestException("The method is not yet implemented");
Review Comment:
nit: UnsupportedOperationException should be there, but nevermind for current PR.
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918501064
##########
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.
Review Comment:
That is the msg taken up from the [KIP](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255070434#KIP932:QueuesforKafka-ListShareGroupOffsetsResult). I see the same is followed in other classes in trunk too ([example](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/DescribeShareGroupsResult.java#L28-L32))
--
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 (1/n) [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918508482
##########
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<>();
+ }
Review Comment:
Makes sense. Will update the PR.
--
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 (1/n) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18571:
URL: https://github.com/apache/kafka/pull/18571#discussion_r1918991457
##########
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.
+ */
Review Comment:
nit: Would be better to write the comments as like javadoc i.e. @return, @param? Same for other method and classes.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsSpec.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+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() {
+ }
+
+ /**
+ * Set the topic partitions whose offsets are to be listed for a share group.
+ */
+ public 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 == null ? Collections.emptyList() : topicPartitions;
Review Comment:
Just check if there are no xlint warnings as `Collections.emptyList()` has unchecked suppression so using List.of() might be better. In case there are no warnings then you can have either.
```suggestion
return topicPartitions == null ? List.of() : topicPartitions;
```
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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);
+ readStateSummaryBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateSummaryResult : combinedResponse.data().results()) {
+ if (readStateSummaryResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateSummaryResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateSummaryBackoff.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 summary RPC for key {}: {}", partitionKey(), error.message());
+ if (!readStateSummaryBackoff.canAttempt()) {
+ log.error("Exhausted max retries for read state summary RPC for key {} without success.", partitionKey());
+ readStateSummaryErrorReponse(error, new Exception("Exhausted max retries to complete read state summary RPC without success."));
+ return;
+ }
+ super.resetCoordinatorNode();
+ timer.add(new PersisterTimerTask(readStateSummaryBackoff.backOff(), this));
+ return;
+
+ default:
+ log.error("Unable to perform read state summary RPC for key {}: {}", partitionKey(), error.message());
+ readStateSummaryErrorReponse(error, null);
+ return;
+ }
+ }
+ }
+ }
+
+ // no response found specific topic partition
+ IllegalStateException exception = new IllegalStateException(
+ "Failed to read state summary 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 read state summary RPC. " +
+ (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 find coordinator. " +
+ (exception == null ? error.message() : exception.getMessage()))));
+ }
Review Comment:
Why do we need method `findCoordinatorErrorResponse` in ReadStateSummaryHandler?
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -34,7 +35,7 @@ public static class Builder extends AbstractRequest.Builder<ReadShareGroupStateS
private final ReadShareGroupStateSummaryRequestData data;
public Builder(ReadShareGroupStateSummaryRequestData data) {
- this(data, false);
+ this(data, true);
Review Comment:
Ok, if it's happening at other places but you can use other constructor as well where you can explicilty pass `true` for `enableUnstableLastVersion` while creating the request builder. Anyways the `enableUnstableLastVersion` for Persister won't be for long hence I leave it on you. But if we default to true then is there even any point of making these RPCs unstable in json?
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListShareGroupOffsetsSpec.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+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() {
+ }
Review Comment:
Should we just remove this default constructor, it's adding no value?
##########
server-common/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 readStateSummaryBackoff;
+
+ 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.readStateSummaryBackoff = 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);
+ readStateSummaryBackoff.incrementAttempt();
+
+ ReadShareGroupStateSummaryResponse combinedResponse = (ReadShareGroupStateSummaryResponse) response.responseBody();
+ for (ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult readStateSummaryResult : combinedResponse.data().results()) {
+ if (readStateSummaryResult.topicId().equals(partitionKey().topicId())) {
+ Optional<ReadShareGroupStateSummaryResponseData.PartitionResult> partitionStateData =
+ readStateSummaryResult.partitions().stream().filter(partitionResult -> partitionResult.partition() == partitionKey().partition())
+ .findFirst();
+
+ if (partitionStateData.isPresent()) {
+ Errors error = Errors.forCode(partitionStateData.get().errorCode());
+ switch (error) {
+ case NONE:
+ readStateSummaryBackoff.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 summary RPC for key {}: {}", partitionKey(), error.message());
+ if (!readStateSummaryBackoff.canAttempt()) {
+ log.error("Exhausted max retries for read state summary RPC for key {} without success.", partitionKey());
+ readStateSummaryErrorReponse(error, new Exception("Exhausted max retries to complete read state summary RPC without success."));
+ return;
+ }
+ super.resetCoordinatorNode();
+ timer.add(new PersisterTimerTask(readStateSummaryBackoff.backOff(), this));
+ return;
+
+ default:
+ log.error("Unable to perform read state summary RPC for key {}: {}", partitionKey(), error.message());
+ readStateSummaryErrorReponse(error, null);
+ return;
+ }
+ }
+ }
+ }
+
+ // no response found specific topic partition
+ IllegalStateException exception = new IllegalStateException(
+ "Failed to read state summary for share partition " + partitionKey()
+ );
+ readStateSummaryErrorReponse(Errors.forException(exception), exception);
Review Comment:
You can just pass `exception` and can convert to Error in the method itself. Don't see any value to pass both explicitly.
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -969,5 +1116,27 @@ private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReads(
.setPartitions(entry.getValue()))
.collect(Collectors.toList())));
}
+
+ private static AbstractRequest.Builder<? extends AbstractRequest> coalesceReadSummarys(String groupId, List<? extends PersisterStateManagerHandler> handlers) {
+ Map<Uuid, List<ReadShareGroupStateSummaryRequestData.PartitionData>> partitionData = new HashMap<>();
+ handlers.forEach(persHandler -> {
Review Comment:
What do we mean by `pers` here?
##########
clients/src/main/java/org/apache/kafka/common/requests/ReadShareGroupStateSummaryRequest.java:
##########
@@ -83,8 +84,24 @@ public ReadShareGroupStateSummaryRequestData data() {
public static ReadShareGroupStateSummaryRequest parse(ByteBuffer buffer, short version) {
return new ReadShareGroupStateSummaryRequest(
- new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
- version
+ new ReadShareGroupStateSummaryRequestData(new ByteBufferAccessor(buffer), version),
+ version
);
}
+
+ public static List<ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult> getErrorList(
+ List<Uuid> topicIds,
+ Errors error
+ ) {
+ return topicIds.stream()
+ .map(topicId ->
+ new ReadShareGroupStateSummaryResponseData.ReadStateSummaryResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(
+ new ReadShareGroupStateSummaryResponseData.PartitionResult()
+ .setErrorCode(error.code())
+ .setErrorMessage(error.message())
Review Comment:
This will give default Error message and original exception message will be lost, do you want that behaviour? If yes, then please write a comment here else please fix it.
--
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