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

Author: PratRanj07

CHANGELOG.md

@@ -39,6 +39,8 @@ librdkafka v2.10.0 is a feature release:
    leader change and offset validation (#4970).
  * Fix the Nagle algorithm (TCP_NODELAY) on broker sockets to not be enabled
    by default (#4986).
+ * [KIP-848] `rd_kafka_DescribeConsumerGroups()` now supports KIP-848 introduced `consumer` groups. Two new fields for consumer group type and target assignment have also been added. Type defines whether this group is a `classic` or `consumer` group. Target assignment is only valid for the `consumer` protocol and its defaults to NULL. (#4922).
+ * [KIP-848] Group Config is now supported in AlterConfigs, IncrementalAlterConfigs and DescribeConfigs. (#4939).
 
 
 ## Fixes
@@ -160,7 +162,6 @@ librdkafka v2.10.0 is a feature release:
 
 librdkafka v2.8.0 is a maintenance release:
 
-* Extend Config Apis to support group config (#4939).
 * Socket options are now all set before connection (#4893).
 * Client certificate chain is now sent when using `ssl.certificate.pem`
   or `ssl_certificate` or `ssl.keystore.location` (#4894).

examples/describe_consumer_groups.c

@@ -164,15 +164,31 @@ print_group_member_info(const rd_kafka_MemberDescription_t *member) {
             rd_kafka_MemberDescription_host(member));
         const rd_kafka_MemberAssignment_t *assignment =
             rd_kafka_MemberDescription_assignment(member);
-        const rd_kafka_topic_partition_list_t *topic_partitions =
+        const rd_kafka_topic_partition_list_t *assigned_topic_partitions =
             rd_kafka_MemberAssignment_partitions(assignment);
-        if (!topic_partitions) {
+        const rd_kafka_MemberAssignment_t *target_assignment =
+            rd_kafka_MemberDescription_target_assignment(member);
+        const rd_kafka_topic_partition_list_t *target_topic_partitions =
+            target_assignment
+                ? rd_kafka_MemberAssignment_partitions(target_assignment)
+                : NULL;
+        if (!assigned_topic_partitions) {
                 printf("    No assignment\n");
-        } else if (topic_partitions->cnt == 0) {
+        } else if (assigned_topic_partitions->cnt == 0) {
                 printf("    Empty assignment\n");
         } else {
                 printf("    Assignment:\n");
-                print_partition_list(stdout, topic_partitions, 0, "      ");
+                print_partition_list(stdout, assigned_topic_partitions, 0,
+                                     "      ");
+        }
+        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,
+                                     "      ");
         }
 }
 
@@ -194,6 +210,8 @@ static void print_group_info(const rd_kafka_ConsumerGroupDescription_t *group) {
             rd_kafka_ConsumerGroupDescription_partition_assignor(group);
         rd_kafka_consumer_group_state_t state =
             rd_kafka_ConsumerGroupDescription_state(group);
+        rd_kafka_consumer_group_type_t type =
+            rd_kafka_ConsumerGroupDescription_type(group);
         authorized_operations =
             rd_kafka_ConsumerGroupDescription_authorized_operations(
                 group, &authorized_operations_cnt);
@@ -212,9 +230,10 @@ static void print_group_info(const rd_kafka_ConsumerGroupDescription_t *group) {
                          rd_kafka_Node_port(coordinator));
         }
         printf(
-            "Group \"%s\", partition assignor \"%s\", "
-            " state %s%s, with %" PRId32 " member(s)\n",
+            "Group \"%s\", partition assignor \"%s\", type \"%s\""
+            " state \"%s\"%s, with %" PRId32 " member(s)\n",
             group_id, partition_assignor,
+            rd_kafka_consumer_group_type_name(type),
             rd_kafka_consumer_group_state_name(state), coordinator_desc,
             member_cnt);
         for (j = 0; j < authorized_operations_cnt; j++) {

src/rdkafka.c

@@ -1690,6 +1690,7 @@ static void rd_kafka_stats_emit_broker_reqs(struct _stats_emit *st,
                 [RD_KAFKAP_AlterClientQuotas]            = rd_true,
                 [RD_KAFKAP_DescribeUserScramCredentials] = rd_true,
                 [RD_KAFKAP_AlterUserScramCredentials]    = rd_true,
+                [RD_KAFKAP_ConsumerGroupDescribe]        = rd_true,
             }};
         int i;
         int cnt = 0;

src/rdkafka.h

@@ -8900,6 +8900,17 @@ RD_EXPORT
 const rd_kafka_Node_t *rd_kafka_ConsumerGroupDescription_coordinator(
     const rd_kafka_ConsumerGroupDescription_t *grpdesc);
 
+/**
+ * @brief Gets type for the \p grpdesc group.
+ *
+ * @param grpdesc The group description.
+ *
+ * @return A group type.
+ */
+RD_EXPORT
+rd_kafka_consumer_group_type_t rd_kafka_ConsumerGroupDescription_type(
+    const rd_kafka_ConsumerGroupDescription_t *grpdesc);
+
 /**
  * @brief Gets the members count of \p grpdesc group.
  *
@@ -9012,6 +9023,21 @@ RD_EXPORT
 const rd_kafka_topic_partition_list_t *rd_kafka_MemberAssignment_partitions(
     const rd_kafka_MemberAssignment_t *assignment);
 
+/**
+ * @brief Gets target assignment of \p member.
+ *
+ * @param member The group member.
+ *
+ * @return The target assignment for `consumer` group types.
+ *         Returns NULL for the `classic` group types.
+ *
+ * @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);
+
 /**@}*/
 
 /**