Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1920964896

   @nizhikov FYI - I merged this one.


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac merged PR #15253:
URL: https://github.com/apache/kafka/pull/15253


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1473247269


##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -158,36 +295,71 @@ public ApiResult handleResponse(
 return new ApiResult<>(completed, failed, new 
ArrayList<>(groupsToUnmap));
 }
 
+private Set 
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+return assignment.topicPartitions().stream().flatMap(topic ->
+topic.partitions().stream().map(partition ->
+new TopicPartition(topic.topicName(), partition)
+)
+).collect(Collectors.toSet());
+}
+
 private void handleError(
 CoordinatorKey groupId,
 Errors error,
+String errorMsg,
 Map failed,
-Set groupsToUnmap
+Set groupsToUnmap,
+boolean isConsumerGroupResponse
 ) {
+String apiName = isConsumerGroupResponse ? "ConsumerGroupDescribe" : 
"DescribeGroups";
+
 switch (error) {
 case GROUP_AUTHORIZATION_FAILED:
-log.debug("`DescribeGroups` request for group id {} failed due 
to error {}", groupId.idValue, error);
-failed.put(groupId, error.exception());
+log.debug("`{}` request for group id {} failed due to error 
{}.", apiName, groupId.idValue, error);
+failed.put(groupId, error.exception(errorMsg));
 break;
 
 case COORDINATOR_LOAD_IN_PROGRESS:
 // If the coordinator is in the middle of loading, then we 
just need to retry
-log.debug("`DescribeGroups` request for group id {} failed 
because the coordinator " +
-"is still in the process of loading state. Will retry", 
groupId.idValue);
+log.debug("`{}` request for group id {} failed because the 
coordinator " +
+"is still in the process of loading state. Will retry.", 
apiName, groupId.idValue);
 break;
 
 case COORDINATOR_NOT_AVAILABLE:
 case NOT_COORDINATOR:
 // If the coordinator is unavailable or there was a 
coordinator change, then we unmap
 // the key so that we retry the `FindCoordinator` request
-log.debug("`DescribeGroups` request for group id {} returned 
error {}. " +
-"Will attempt to find the coordinator again and retry", 
groupId.idValue, error);
+log.debug("`{}` request for group id {} returned error {}. " +
+"Will attempt to find the coordinator again and retry.", 
apiName, groupId.idValue, error);
 groupsToUnmap.add(groupId);
 break;
 
+case UNSUPPORTED_VERSION:
+if (isConsumerGroupResponse) {
+log.debug("`{}` request for group id {} failed because the 
API is not " +
+"supported. Will retry with `DescribeGroups` API.", 
apiName, groupId.idValue);
+useClassicGroupApi.add(groupId.idValue);
+} else {
+log.error("`{}` request for group id {} because the 
`ConsumerGroupDescribe` API is not supported.",
+apiName, groupId.idValue);
+failed.put(groupId, error.exception(errorMsg));
+}
+break;
+
+case GROUP_ID_NOT_FOUND:
+if (isConsumerGroupResponse) {
+log.debug("`{}` request for group id {} failed because the 
group is not " +
+"a new consumer group. Will retry with 
`DescribeGroups` API.", apiName, groupId.idValue);
+useClassicGroupApi.add(groupId.idValue);
+} else {
+log.error("`{}` request for group id {} because the group 
does not exist.", apiName, groupId.idValue);

Review Comment:
   Good catch! Let me fix this.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

AndrewJSchofield commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1473234152


##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -158,36 +295,71 @@ public ApiResult handleResponse(
 return new ApiResult<>(completed, failed, new 
ArrayList<>(groupsToUnmap));
 }
 
+private Set 
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+return assignment.topicPartitions().stream().flatMap(topic ->
+topic.partitions().stream().map(partition ->
+new TopicPartition(topic.topicName(), partition)
+)
+).collect(Collectors.toSet());
+}
+
 private void handleError(
 CoordinatorKey groupId,
 Errors error,
+String errorMsg,
 Map failed,
-Set groupsToUnmap
+Set groupsToUnmap,
+boolean isConsumerGroupResponse
 ) {
+String apiName = isConsumerGroupResponse ? "ConsumerGroupDescribe" : 
"DescribeGroups";
+
 switch (error) {
 case GROUP_AUTHORIZATION_FAILED:
-log.debug("`DescribeGroups` request for group id {} failed due 
to error {}", groupId.idValue, error);
-failed.put(groupId, error.exception());
+log.debug("`{}` request for group id {} failed due to error 
{}.", apiName, groupId.idValue, error);
+failed.put(groupId, error.exception(errorMsg));
 break;
 
 case COORDINATOR_LOAD_IN_PROGRESS:
 // If the coordinator is in the middle of loading, then we 
just need to retry
-log.debug("`DescribeGroups` request for group id {} failed 
because the coordinator " +
-"is still in the process of loading state. Will retry", 
groupId.idValue);
+log.debug("`{}` request for group id {} failed because the 
coordinator " +
+"is still in the process of loading state. Will retry.", 
apiName, groupId.idValue);
 break;
 
 case COORDINATOR_NOT_AVAILABLE:
 case NOT_COORDINATOR:
 // If the coordinator is unavailable or there was a 
coordinator change, then we unmap
 // the key so that we retry the `FindCoordinator` request
-log.debug("`DescribeGroups` request for group id {} returned 
error {}. " +
-"Will attempt to find the coordinator again and retry", 
groupId.idValue, error);
+log.debug("`{}` request for group id {} returned error {}. " +
+"Will attempt to find the coordinator again and retry.", 
apiName, groupId.idValue, error);
 groupsToUnmap.add(groupId);
 break;
 
+case UNSUPPORTED_VERSION:
+if (isConsumerGroupResponse) {
+log.debug("`{}` request for group id {} failed because the 
API is not " +
+"supported. Will retry with `DescribeGroups` API.", 
apiName, groupId.idValue);
+useClassicGroupApi.add(groupId.idValue);
+} else {
+log.error("`{}` request for group id {} because the 
`ConsumerGroupDescribe` API is not supported.",
+apiName, groupId.idValue);
+failed.put(groupId, error.exception(errorMsg));
+}
+break;
+
+case GROUP_ID_NOT_FOUND:
+if (isConsumerGroupResponse) {
+log.debug("`{}` request for group id {} failed because the 
group is not " +
+"a new consumer group. Will retry with 
`DescribeGroups` API.", apiName, groupId.idValue);
+useClassicGroupApi.add(groupId.idValue);
+} else {
+log.error("`{}` request for group id {} because the group 
does not exist.", apiName, groupId.idValue);

Review Comment:
   This doesn't seem grammatical. It would say "DescribeGroups request for 
group id {} because the group does not exist". I suggest "failed" or something 
like that is missing.



##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -158,36 +295,71 @@ public ApiResult handleResponse(
 return new ApiResult<>(completed, failed, new 
ArrayList<>(groupsToUnmap));
 }
 
+private Set 
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+return assignment.topicPartitions().stream().flatMap(topic ->
+topic.partitions().stream().map(partition ->
+new TopicPartition(topic.topicName(), partition)
+)
+).collect(Collectors.toSet());
+}
+
 private void handleError(
 CoordinatorKey groupId,
 Errors error,
+String errorMsg,
 Map failed,
-Set groupsToUnmap
+ 

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1919372877

   Thanks @cadonna! I have addressed your comments.


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

cadonna commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1472830748


##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -89,18 +97,42 @@ public AdminApiLookupStrategy 
lookupStrategy() {
 }
 
 @Override
-public DescribeGroupsRequest.Builder buildBatchedRequest(int 
coordinatorId, Set keys) {
-List groupIds = keys.stream().map(key -> {
+public Collection> buildRequest(int 
coordinatorId, Set keys) {
+Set newConsumerGroups = new HashSet<>();
+Set oldConsumerGroups = new HashSet<>();
+
+keys.forEach(key -> {
 if (key.type != FindCoordinatorRequest.CoordinatorType.GROUP) {
-throw new IllegalArgumentException("Invalid transaction 
coordinator key " + key +
+throw new IllegalArgumentException("Invalid group coordinator 
key " + key +
 " when building `DescribeGroups` request");
 }
-return key.idValue;
-}).collect(Collectors.toList());
-DescribeGroupsRequestData data = new DescribeGroupsRequestData()
-.setGroups(groupIds)
-.setIncludeAuthorizedOperations(includeAuthorizedOperations);
-return new DescribeGroupsRequest.Builder(data);
+
+// Be default, we always try with using the new consumer group

Review Comment:
   typo:
   ```suggestion
   // By default, we always try using the new consumer group
   ```



##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -51,11 +57,12 @@
 import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
 
-public class DescribeConsumerGroupsHandler extends 
AdminApiHandler.Batched {
+public class DescribeConsumerGroupsHandler implements 
AdminApiHandler {
 
 private final boolean includeAuthorizedOperations;
 private final Logger log;
 private final AdminApiLookupStrategy lookupStrategy;
+private final Map useClassicGroupApi;

Review Comment:
   I am wondering whether it would be enough to use a set here instead of a 
map. As far as I can see you always put the value `true` into the map. So 
either the key exists with value `true` or it does not exist. The case that a 
key exists with value `false` does not seem to occur with this code.



##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -158,36 +297,67 @@ public ApiResult handleResponse(
 return new ApiResult<>(completed, failed, new 
ArrayList<>(groupsToUnmap));
 }
 
+private Set 
convertAssignment(ConsumerGroupDescribeResponseData.Assignment assignment) {
+return assignment.topicPartitions().stream().flatMap(topic ->
+topic.partitions().stream().map(partition ->
+new TopicPartition(topic.topicName(), partition)
+)
+).collect(Collectors.toSet());
+}
+
 private void handleError(
 CoordinatorKey groupId,
 Errors error,
+String errorMsg,
 Map failed,
-Set groupsToUnmap
+Set groupsToUnmap,
+boolean isConsumerGroupResponse
 ) {
 switch (error) {
 case GROUP_AUTHORIZATION_FAILED:
-log.debug("`DescribeGroups` request for group id {} failed due 
to error {}", groupId.idValue, error);
-failed.put(groupId, error.exception());
+log.debug("`DescribeGroups` request for group id {} failed due 
to error {}.", groupId.idValue, error);
+failed.put(groupId, error.exception(errorMsg));
 break;
 
 case COORDINATOR_LOAD_IN_PROGRESS:
 // If the coordinator is in the middle of loading, then we 
just need to retry
 log.debug("`DescribeGroups` request for group id {} failed 
because the coordinator " +
-"is still in the process of loading state. Will retry", 
groupId.idValue);
+"is still in the process of loading state. Will retry.", 
groupId.idValue);
 break;
 
 case COORDINATOR_NOT_AVAILABLE:
 case NOT_COORDINATOR:
 // If the coordinator is unavailable or there was a 
coordinator change, then we unmap
 // the key so that we retry the `FindCoordinator` request
 log.debug("`DescribeGroups` request for group id {} returned 
error {}. " +
-"Will attempt to find the coordinator again and retry", 
groupId.idValue, error);
+"Will attempt to find the coordinator again and retry.", 
groupId.idValue, error);
 groupsToUnmap.add(groupId);
 break;
 
+case UNSUPPORTED_VERSION:
+if (isConsumerGroupResponse) {
+log.debug("`DescribeGroups` request for 

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1916713439

   Hey @mimaison! As you have done some work in this area, I wonder if you 
would be interested by reviewing this PR. It would help us with the KIP-848 
effort. We need this as soon as possible for testing purposes. Would you have 
time?


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1471105525


##
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java:
##
@@ -36,6 +37,7 @@ public class ConsumerGroupDescription {
 private final boolean isSimpleConsumerGroup;
 private final Collection members;
 private final String partitionAssignor;
+private final GroupType type;

Review Comment:
   We have a String here in the KIP. It seems better to use an Enum as we have 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1471064233


##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##
@@ -240,10 +240,11 @@ private CompletableFuture maybeAutoCommit(final 
Map result = addOffsetCommitRequest(offsets, 
expirationTimeMs, retryOnStaleEpoch)
-.whenComplete(autoCommitCallback(offsets));
 autocommit.resetTimer();
 autocommit.setInflightCommitStatus(true);
+

Review Comment:
   We have raised a PR to fix this bug: 
https://github.com/apache/kafka/pull/15282. We can ignore it 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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1915007420

   Thanks for letting me know, @nizhikov. I can definitely adapt my PR when 
https://github.com/apache/kafka/pull/15256 gets merged. However, I cannot wait 
too long because this PR is needed for testing KIP-848.


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

nizhikov commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1914941776

   Hello @dajac 
   
   Right now I have #15256 which moves `DescribeConsumerGroupTest` and 
`ConsumerGroupCommandTest` to tools module.
   You are modifying those tests in PR.
   My PR is ready and I hope will be merged soone.
   
   May be you can wait for it and modify tests in `tools`? 
   Or even review it, you may be familiar with the tests right now


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on PR #15253:
URL: https://github.com/apache/kafka/pull/15253#issuecomment-1912511055

   cc @nizhikov @jolshan FYI - I have a few minor changes related to the 
consumer group command here. As saw that you’re working on migrating it to Java.


-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] KAFKA-16189; Extend admin to support ConsumerGroupDescribe API [kafka]

[via GitHub]

dajac commented on code in PR #15253:
URL: https://github.com/apache/kafka/pull/15253#discussion_r1465170111


##
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java:
##
@@ -89,18 +96,42 @@ public AdminApiLookupStrategy 
lookupStrategy() {
 }
 
 @Override
-public DescribeGroupsRequest.Builder buildBatchedRequest(int 
coordinatorId, Set keys) {
-List groupIds = keys.stream().map(key -> {
+public Collection> buildRequest(int 
coordinatorId, Set keys) {
+Set newConsumerGroups = new HashSet<>();
+Set oldConsumerGroups = new HashSet<>();
+
+keys.forEach(key -> {
 if (key.type != FindCoordinatorRequest.CoordinatorType.GROUP) {
-throw new IllegalArgumentException("Invalid transaction 
coordinator key " + key +
+throw new IllegalArgumentException("Invalid group coordinator 
key " + key +
 " when building `DescribeGroups` request");
 }
-return key.idValue;
-}).collect(Collectors.toList());
-DescribeGroupsRequestData data = new DescribeGroupsRequestData()
-.setGroups(groupIds)
-.setIncludeAuthorizedOperations(includeAuthorizedOperations);
-return new DescribeGroupsRequest.Builder(data);
+
+// Be default, we always try with using the new consumer group
+// describe API. If it fails, we fail back to using the classic
+// group API.
+if (useClassicGroupApi.getOrDefault(key.idValue, false)) {
+oldConsumerGroups.add(key);
+} else {
+newConsumerGroups.add(key);
+}
+});
+
+List> requests = new ArrayList<>();
+if (!newConsumerGroups.isEmpty()) {
+ConsumerGroupDescribeRequestData data = new 
ConsumerGroupDescribeRequestData()
+.setGroupIds(newConsumerGroups.stream().map(key -> 
key.idValue).collect(Collectors.toList()))
+.setIncludeAuthorizedOperations(includeAuthorizedOperations);
+requests.add(new RequestAndKeys<>(new 
ConsumerGroupDescribeRequest.Builder(data, true), newConsumerGroups));

Review Comment:
   `true` must be removed here when https://github.com/apache/kafka/pull/15255 
is merged.



-- 
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-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org