KIP 848: Added support for DescribeConsumerGroup for consumer protocol groups #4941

PratRanj07 · 2024-12-27T11:22:46Z

This PR intends to add support for calling DescribeConsumerGroups for groups created with the new consumer protocol and add backward compatibility to describe group created with the old Classic Protocol.
The working of the API remains similar. Users will continue to call the existing DescribeConsumerGroups api but internally we will be calling the new ConsumerGroupDescribe api to fetch the results. If we get GRP_NOT_FOUND or UNSUPPORTED_VERSION error, then in that case we will call the old DescribeConsumerGroup api.

confluent-cla-assistant · 2024-12-27T11:22:56Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

pranavrth

First pass comments.

pranavrth · 2025-02-04T08:28:42Z

src/rdkafka.h

+/**
+ * @brief Gets target assignment of \p member.
+ *
+ * @param member The group member.
+ *
+ * @return The target assignment.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p member object.
+ */
+RD_EXPORT
+const rd_kafka_MemberAssignment_t *rd_kafka_MemberDescription_target_assignment(
+    const rd_kafka_MemberDescription_t *member);
+
+/**
+ * @brief Gets target assigned partitions of a member \p assignment.
+ *
+ * @param assignment The group member assignment.
+ *
+ * @return The target assigned partitions.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p assignment object.
+ */
+RD_EXPORT
+const rd_kafka_topic_partition_list_t *
+rd_kafka_MemberAssignment_target_partitions(
+    const rd_kafka_MemberAssignment_t *assignment);
+


What are we doing for the old protocol on these new functions?

It is giving NULL as default value in these cases. I have updated it in the comments

pranavrth · 2025-02-06T07:25:51Z

CHANGELOG.md

@@ -2,6 +2,7 @@

 librdkafka v2.8.0 is a maintenance release:

+* DescribeConsumerGroup now supports new consumer protocol groups (#4922).


Add that new fields type and target assignment is added. Target assignment is only applicable for Consumer protocol. Explain about the default values as well.

pranavrth · 2025-02-06T07:27:54Z

examples/describe_consumer_groups.c

+        const rd_kafka_topic_partition_list_t *target_topic_partitions =
+            rd_kafka_MemberAssignment_partitions(target_assignment);


This will give NULL pointer exception as target_assignment is only applicable for consumer protocol. It should be NULL for classic protocol and default value should be NULL.

pranavrth · 2025-02-06T07:28:56Z

examples/describe_consumer_groups.c

+        if (!target_topic_partitions) {
+                printf("    No target assignment\n");
+        } else if (target_topic_partitions->cnt == 0) {
+                printf("    Empty target assignment\n");
+        } else {
+                printf("    Target assignment:\n");
+                print_partition_list(stdout, target_topic_partitions, 0,
+                                     "      ");
+        }


This is different for classic protocol.

pranavrth · 2025-02-06T07:39:52Z

src/rdkafka_proto.h

-            [RD_KAFKAP_ListTransactions]          = "ListTransactions",
-            [RD_KAFKAP_AllocateProducerIds]       = "AllocateProducerIds",
-            [RD_KAFKAP_ConsumerGroupHeartbeat]    = "ConsumerGroupHeartbeat",
+            [RD_KAFKAP_ConsumerGroupDescribe]  = "ConsumerGroupDescribeRequest",


Better add this at the end.

pranavrth · 2025-02-06T08:07:21Z

src/rdkafka_request.c

+ */
+rd_kafka_resp_err_t
+rd_kafka_ConsumerGroupDescribeRequest(rd_kafka_broker_t *rkb,
+                                      const rd_list_t *groups /*(char*)*/,


This should be char ** and accept count as well. Check older API.

pranavrth · 2025-02-06T08:28:59Z

src/rdkafka_request.c

+                                      rd_kafka_resp_cb_t *resp_cb,
+                                      void *opaque) {
+        rd_kafka_buf_t *rkbuf;
+        int16_t maxApiVersion = 0;


No need for this variable. Used only once.

pranavrth · 2025-02-06T08:29:42Z

src/rdkafka_request.c

+        int16_t ApiVersion    = rd_kafka_broker_ApiVersion_supported(
+            rkb, RD_KAFKAP_ConsumerGroupDescribe, 0, maxApiVersion, NULL);
+        size_t ofGroupsArrayCnt;
+        int grp_ids_cnt = rd_list_cnt(groups);
+        int i, include_authorized_operations;
+        char *group;


Suggested change

int16_t ApiVersion = rd_kafka_broker_ApiVersion_supported(

rkb, RD_KAFKAP_ConsumerGroupDescribe, 0, maxApiVersion, NULL);

size_t ofGroupsArrayCnt;

int grp_ids_cnt = rd_list_cnt(groups);

int i, include_authorized_operations;

char *group;

size_t ofGroupsArrayCnt;

int grp_ids_cnt = rd_list_cnt(groups);

int i, include_authorized_operations;

char *group;

int16_t ApiVersion = rd_kafka_broker_ApiVersion_supported(

rkb, RD_KAFKAP_ConsumerGroupDescribe, 0, maxApiVersion, NULL);

pranavrth · 2025-02-06T17:20:44Z

src/rdkafka_request.c

+        rkbuf = rd_kafka_buf_new_flexver_request(
+            rkb, RD_KAFKAP_ConsumerGroupDescribe, 1,
+            4 /* rd_kafka_buf_write_arraycnt_pos */ +
+                1 /* IncludeAuthorizedOperations */ + 1 /* tags */ +
+                32 * grp_ids_cnt /* Groups */,
+            rd_true /* flexver */);


pranavrth · 2025-02-07T07:29:31Z

src/rdkafka_admin.c

+        if (newgroupres->type == RD_KAFKA_CONSUMER_GROUP_TYPE_CONSUMER &&
+            newgroupres->error &&
+            (newgroupres->error->code == RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND ||
+             newgroupres->error->code ==
+                 RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION)) {


Check for memory leak in this condition as well.

pranavrth

Still reviewing. Few comments. Check for possible SegFaults.

pranavrth · 2025-02-14T07:06:06Z

src/rdkafka_admin.c

-                                              src->assignment.partitions);
+        return rd_kafka_MemberDescription_new(
+            src->client_id, src->consumer_id, src->group_instance_id, src->host,
+            src->assignment.partitions, src->target_assignment.partitions);


This will give a SegFault if target assignment is NULL

pranavrth · 2025-02-14T07:06:30Z

src/rdkafka_admin.c

+        if (member->target_assignment.partitions)
+                rd_kafka_topic_partition_list_destroy(
+                    member->target_assignment.partitions);
        rd_free(member);


Same changes of SegFault.

pranavrth · 2025-02-14T07:18:34Z

src/rdkafka_admin.c

+        RD_LIST_FOREACH(group, groups, i) {
+                groups_arr[i] = rd_list_elem(groups, i);
+        }
+


No need for this. You can directly use groups->rl_elems

PratRanj07 added 9 commits December 27, 2024 16:47

first pass

f3c7259

debugging

5246d91

API working now

0d16888

Api working now, tests left

041a306

Wrote tests

56dcce6

style fix

f4adaf8

style fix

01a1628

removing unused variables

5488ec7

trying to fix build

4863104

PratRanj07 requested a review from a team as a code owner December 27, 2024 11:22

PratRanj07 added 2 commits December 27, 2024 16:59

changelog change

71d0236

Merge branch 'master' into dev_ConsumerGroupDescribeApi

ecb6ff8

pranavrth requested changes Feb 7, 2025

View reviewed changes

PratRanj07 added 5 commits February 10, 2025 12:36

Merge Master

812f0fc

Requested changes

6ba8777

var change

95e871f

style fix

1554a2b

style fix

0293397

PratRanj07 requested a review from pranavrth February 12, 2025 11:38

pranavrth requested changes Feb 14, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KIP 848: Added support for DescribeConsumerGroup for consumer protocol groups #4941

KIP 848: Added support for DescribeConsumerGroup for consumer protocol groups #4941

PratRanj07 commented Dec 27, 2024

confluent-cla-assistant bot commented Dec 27, 2024

pranavrth left a comment

pranavrth Feb 4, 2025

PratRanj07 Feb 12, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 6, 2025

pranavrth Feb 7, 2025

pranavrth left a comment

pranavrth Feb 14, 2025

pranavrth Feb 14, 2025

pranavrth Feb 14, 2025

		@@ -2,6 +2,7 @@

		librdkafka v2.8.0 is a maintenance release:

		* DescribeConsumerGroup now supports new consumer protocol groups (#4922).

		const rd_kafka_topic_partition_list_t *target_topic_partitions =
		rd_kafka_MemberAssignment_partitions(target_assignment);

KIP 848: Added support for DescribeConsumerGroup for consumer protocol groups #4941

Are you sure you want to change the base?

KIP 848: Added support for DescribeConsumerGroup for consumer protocol groups #4941

Conversation

PratRanj07 commented Dec 27, 2024

confluent-cla-assistant bot commented Dec 27, 2024

pranavrth left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pranavrth left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment