KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config #4939

PratRanj07 · 2024-12-19T11:12:28Z

This PR intends to add support for Group Resource type for IncrementalALterConfigs API as specified in KIP 848 and extended the DescribeConfigs support till API version 3.
Additional changed the integration test to check for Group Resource Type.

confluent-cla-assistant · 2024-12-19T11:12:39Z

🎉 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 high level pass

pranavrth · 2024-12-26T08:14:27Z

src/rdkafka_admin.c

@@ -2911,6 +2921,47 @@ const char *rd_kafka_ResourceType_name(rd_kafka_ResourceType_t restype) {
 }


+rd_kafka_InternalConfigResourceType_t
+map_to_internal_config_resourcetype(rd_kafka_ResourceType_t resourcetype) {


name both consistently. map_from_x_to_y

pranavrth · 2024-12-26T08:14:32Z

src/rdkafka_admin.c

+        }
+}
+
+rd_kafka_ResourceType_t map_from_internal_config_resourcetype(


name both consistently. map_from_x_to_y

pranavrth · 2024-12-26T08:23:52Z

src/rdkafka_admin.c

+        default:
+                return RD_KAFKA_INTERNAL_RESOURCE_CONFIG_UNKNOWN;


Don't make anything default. Add an assert instead that the value is not supported.

pranavrth · 2024-12-26T08:24:02Z

src/rdkafka_admin.c

+        default:
+                return RD_KAFKA_RESOURCE_UNKNOWN;


pranavrth · 2024-12-26T08:25:30Z

src/rdkafka_admin.c

+                        if (api_version >= 3) {
+                                rd_kafka_buf_read_i8(reply, &config_type);
+                                rd_kafka_buf_read_str(reply, &documentation);
+                                entry->type = config_type;
+                                entry->documentation =
+                                    RD_KAFKAP_STR_DUP(&documentation);
+                        }


Is this change necessary for KIP-848 changes?

No this change was for upgrading the API version. Have removed it.

pranavrth · 2024-12-26T08:26:28Z

src/rdkafka_admin.c

@@ -3805,6 +3880,7 @@ rd_kafka_DescribeConfigsResponse_parse(rd_kafka_op_t *rko_req,
                            config);
                config = NULL;
        }
+        rd_kafka_buf_skip_tags(reply);


pranavrth · 2024-12-26T08:31:28Z

src/rdkafka_request.c

                /* include_synonyms */
                rd_kafka_buf_write_i8(rkbuf, 1);
        }

+        if (ApiVersion >= 3) {
+                /* include_documentation */
+                rd_kafka_buf_write_i8(rkbuf, 1);


Why always true?

removed it since wasn't neccesarry as we are not upgrading the api version

pranavrth · 2024-12-26T08:32:24Z

src/rdkafka_request.c

@@ -5445,7 +5447,7 @@ rd_kafka_resp_err_t rd_kafka_DescribeConfigsRequest(
        }

        ApiVersion = rd_kafka_broker_ApiVersion_supported(
-            rkb, RD_KAFKAP_DescribeConfigs, 0, 1, NULL);
+            rkb, RD_KAFKAP_DescribeConfigs, 0, 3, NULL);


I would suggest not to increase the version if it is not necessary. We would need to improve APIs as well to support that and there would be bunch of other things as well which we should be taking care of.

pranavrth · 2024-12-26T08:33:28Z

src/rdkafka_request.c

+                /* include_documentation */
+                rd_kafka_buf_write_i8(rkbuf, 1);
+        }
+        rd_kafka_buf_write_tags_empty(rkbuf);


No need to write tags if we are not supporting flexver.

pranavrth · 2024-12-26T08:33:41Z

src/rdkafka_request.c

@@ -5484,14 +5489,21 @@ rd_kafka_resp_err_t rd_kafka_DescribeConfigsRequest(
                        /* config_name */
                        rd_kafka_buf_write_str(rkbuf, entry->kv->name, -1);
                }
+                rd_kafka_buf_write_tags_empty(rkbuf);


pranavrth

Please do the changes for the comments which are valid now.

pranavrth · 2024-12-26T11:32:01Z

src/rdkafka_admin.c

+        if (resourcetype > RD_KAFKA_RESOURCE__CNT) {
+                rd_assert("Invalid resource type");
+        }


make this default case in the below switch.

pranavrth · 2024-12-26T11:37:42Z

src/rdkafka_admin.c

+    rd_kafka_ResourceType_t resourcetype) {
+
+        if (resourcetype > RD_KAFKA_RESOURCE__CNT) {
+                rd_assert("Invalid resource type");


Provide information about the resource type value in the assert as well.

pranavrth · 2024-12-26T11:37:57Z

src/rdkafka_admin.c

+
+        if (internal_resourcetype != RD_KAFKA_INTERNAL_RESOURCE_CONFIG_GROUP &&
+            internal_resourcetype > RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT) {
+                rd_assert("Recieved invalid resource type");


Provide information about the resource type value in the assert as well.

pranavrth · 2024-12-26T11:38:39Z

src/rdkafka_admin.c

+        case RD_KAFKA_RESOURCE__CNT:
+                return RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT;


We don't want cnt conversion. This is incorrect in my opinion.

pranavrth · 2024-12-26T11:38:55Z

src/rdkafka_admin.c

+        case RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT:
+                return RD_KAFKA_RESOURCE__CNT;


Count conversion is not needed.

pranavrth · 2024-12-26T11:43:30Z

src/rdkafka_admin.h

@@ -283,6 +283,33 @@ struct rd_kafka_ConfigResource_result_s {
                              *   but with response error values. */
 };

+typedef enum rd_kafka_InternalConfigResourceType_t {


Don't use Internal with ConfigResource. Use Java name which is rd_kafka_ConfigResource_t

pranavrth · 2024-12-26T12:22:31Z

src/rdkafka_admin.h

+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_UNKNOWN = 0,
+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_ANY     = 1,
+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_TOPIC   = 2,
+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_GROUP =
+            32,  // Changed value for config APIs
+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_BROKER = 4,
+        RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT,
+} rd_kafka_InternalConfigResourceType_t;


Let's do this conversion only for GROUP right now. We can think about others later.

pranavrth

Tests next.

pranavrth · 2024-12-26T13:33:32Z

src/rdkafka_admin.c

+rd_kafka_ConfigResourceType_t map_from_resource_type_to_config_resource_type(
+    rd_kafka_ResourceType_t resourcetype) {


Suggested change

rd_kafka_ConfigResourceType_t map_from_resource_type_to_config_resource_type(

rd_kafka_ResourceType_t resourcetype) {

int8_t map_from_resource_type_to_config_resource_type(

rd_kafka_ResourceType_t restype) {

pranavrth · 2024-12-26T13:34:11Z

src/rdkafka_admin.c

+rd_kafka_ResourceType_t map_from_config_resource_type_to_resource_type(
+    rd_kafka_ConfigResourceType_t internal_resourcetype) {


Suggested change

rd_kafka_ResourceType_t map_from_config_resource_type_to_resource_type(

rd_kafka_ConfigResourceType_t internal_resourcetype) {

rd_kafka_ResourceType_t map_from_config_resource_type_to_resource_type(

int8_t config_resource_type) {

pranavrth · 2024-12-26T13:34:49Z

src/rdkafka_admin.c

+        if (resourcetype == RD_KAFKA_RESOURCE_GROUP) {
+                return RD_KAFKA_CONFIG_RESOURCE_GROUP;
+        }


Suggested change

if (resourcetype == RD_KAFKA_RESOURCE_GROUP) {

return RD_KAFKA_CONFIG_RESOURCE_GROUP;

}

if (resourcetype == RD_KAFKA_RESOURCE_GROUP)

return RD_KAFKA_CONFIG_RESOURCE_GROUP;

pranavrth · 2024-12-26T13:35:02Z

src/rdkafka_admin.c

+        if (internal_resourcetype == RD_KAFKA_CONFIG_RESOURCE_GROUP) {
+                return RD_KAFKA_RESOURCE_GROUP;
+        }


Suggested change

if (internal_resourcetype == RD_KAFKA_CONFIG_RESOURCE_GROUP) {

return RD_KAFKA_RESOURCE_GROUP;

}

if (internal_resourcetype == RD_KAFKA_CONFIG_RESOURCE_GROUP)

return RD_KAFKA_RESOURCE_GROUP;

pranavrth · 2024-12-26T13:37:18Z

src/rdkafka_admin.c

@@ -3377,11 +3397,14 @@ rd_kafka_IncrementalAlterConfigsResponse_parse(rd_kafka_op_t *rko_req,

                rd_kafka_buf_read_i16(reply, &error_code);
                rd_kafka_buf_read_str(reply, &error_msg);
-                rd_kafka_buf_read_i8(reply, &res_type);
+                rd_kafka_buf_read_i8(reply, &internal_res_type);


config_resource_type

pranavrth · 2024-12-26T13:37:30Z

src/rdkafka_admin.c

@@ -3638,6 +3661,7 @@ rd_kafka_DescribeConfigsResponse_parse(rd_kafka_op_t *rko_req,
        for (i = 0; i < (int)res_cnt; i++) {
                int16_t error_code;
                rd_kafkap_str_t error_msg;
+                int8_t internal_res_type;


config_resource_type

pranavrth · 2024-12-26T13:38:33Z

src/rdkafka_admin.h

+typedef enum rd_kafka_ConfigResourceType_t {
+        RD_KAFKA_CONFIG_RESOURCE_GROUP = 32,  // Changed value for config APIs
+} rd_kafka_ConfigResourceType_t;
+
+/**
+ * @brief Map from rd_kafka_ResourceType_t to
+ * rd_kafka_ConfigResourceType_t
+ */
+rd_kafka_ConfigResourceType_t map_from_resource_type_to_config_resource_type(
+    rd_kafka_ResourceType_t resourcetype);
+
+/**
+ * @brief Map from rd_kafka_ConfigResourceType_t to
+ * rd_kafka_ResourceType_t
+ */
+
+rd_kafka_ResourceType_t map_from_config_resource_type_to_resource_type(
+    rd_kafka_ConfigResourceType_t internal_resourcetype);
+
+


same name, variable and type changes here as well.

pranavrth · 2024-12-26T13:46:00Z

src/rdkafka_admin.h

@@ -283,6 +283,26 @@ struct rd_kafka_ConfigResource_result_s {
                              *   but with response error values. */
 };

+typedef enum rd_kafka_ConfigResourceType_t {
+        RD_KAFKA_CONFIG_RESOURCE_GROUP = 32,  // Changed value for config APIs


Suggested change

RD_KAFKA_CONFIG_RESOURCE_GROUP = 32, // Changed value for config APIs

RD_KAFKA_CONFIG_RESOURCE_GROUP = 32,

pranavrth · 2024-12-26T15:08:34Z

src/rdkafka_admin.h

+/**
+ * @brief Map from int8_t to rd_kafka_ResourceType_t
+ */
+


Remove extra line.

pranavrth · 2024-12-26T15:08:44Z

src/rdkafka_admin.h

@@ -283,6 +283,24 @@ struct rd_kafka_ConfigResource_result_s {
                              *   but with response error values. */
 };

+typedef enum rd_kafka_ConfigResourceType_t {


pranavrth · 2024-12-26T15:19:52Z

tests/0081-admin.c

+        if (!test_consumer_group_protocol_classic()) {
+                exp_err[ci] = RD_KAFKA_RESP_ERR_NO_ERROR;
+        } else {
+                exp_err[ci] = RD_KAFKA_RESP_ERR_INVALID_REQUEST;
+        }


We should check broker version as well here.

pranavrth · 2024-12-26T15:22:35Z

tests/0081-admin.c

+        if (!test_consumer_group_protocol_classic()) {
+                exp_err[ci] = RD_KAFKA_RESP_ERR_NO_ERROR;
+        } else {
+                exp_err[ci] = RD_KAFKA_RESP_ERR_INVALID_REQUEST;
+        }


pranavrth · 2024-12-26T15:25:14Z

tests/0081-admin.c

        do_test_DescribeConfigs(rk, mainq);
+        do_test_DescribeConfigs_groups(rk, mainq);


Why not add new condition in do_test_DescribeConfigs?

pranavrth · 2024-12-26T15:42:16Z

src/rdkafka_admin.h

There is a problem that the error message contain 32 instead of 3. We are not exposing 32 to the user. I think we should modify the error code in the response as well though it is difficult to catch all the cases. Can be an improvement though.

ConfigResource result: 3,my-group: error: Unknown resource type 32

pranavrth · 2024-12-27T09:15:26Z

src/rdkafka_admin.h

+} rd_kafka_ConfigResourceType_t;
+
+/**
+ * @brief Map from rd_kafka_ResourceType_t to int8_t


Better doc explaining what and why are we doing this.

pranavrth · 2024-12-27T09:15:30Z

src/rdkafka_admin.h

+map_from_resource_type_to_config_resource_type(rd_kafka_ResourceType_t restype);
+
+/**
+ * @brief Map from int8_t to rd_kafka_ResourceType_t


Better doc explaining what and why are we doing this.

pranavrth · 2024-12-27T09:22:25Z

tests/0081-admin.c

@@ -947,6 +948,10 @@ static void do_test_IncrementalAlterConfigs(rd_kafka_t *rk,
                const char *confs_delete_subtract_broker[] = {
                    "background.threads", "DELETE",   "",
                    "log.cleanup.policy", "SUBTRACT", "compact"};
+                const char *confs_set_append_group[] = {


Suggested change

const char *confs_set_append_group[] = {

const char *confs_set_group[] = {

pranavrth · 2024-12-27T09:40:20Z

tests/0081-admin.c

+                                rd_kafka_ConfigResource_type(rconfigs[i])),
+                            rd_kafka_ConfigResource_name(rconfigs[i]));
+                        fails++;
+                        continue;


continue; not needed.

PratRanj07 added 2 commits December 19, 2024 15:48

Added group support for config apis

e1d0df0

style fix

8caa707

PratRanj07 requested a review from a team as a code owner December 19, 2024 11:12

PratRanj07 added 2 commits December 19, 2024 21:51

Removed line

8bc4cd1

Removed line

f37bf36

This was referenced Dec 19, 2024

KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config confluentinc/confluent-kafka-go#1344

Open

KIP 848: Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config confluentinc/confluent-kafka-dotnet#2366

Open

PratRanj07 changed the title ~~Add Group Support for Config APIs~~ KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config Dec 19, 2024

PratRanj07 mentioned this pull request Dec 19, 2024

KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config confluentinc/confluent-kafka-python#1856

Open

pranavrth requested changes Dec 26, 2024

View reviewed changes

PratRanj07 added 2 commits December 26, 2024 14:59

removing api ver 3 changes

a11e9bf

Adding assertion and removing default condition

2be7858

PratRanj07 requested a review from pranavrth December 26, 2024 10:27

pranavrth requested changes Dec 26, 2024

View reviewed changes

PratRanj07 added 2 commits December 26, 2024 18:06

new changes

1f5d439

name changes

d05781e

PratRanj07 requested a review from pranavrth December 26, 2024 12:43

pranavrth requested changes Dec 26, 2024

View reviewed changes

Name changes

a4dd049

PratRanj07 requested a review from pranavrth December 26, 2024 14:04

pranavrth requested changes Dec 26, 2024

View reviewed changes

pranavrth reviewed Dec 26, 2024

View reviewed changes

new changes

527c4a3

pranavrth requested changes Dec 27, 2024

View reviewed changes

PratRanj07 added 3 commits December 27, 2024 15:31

requested changes

d42ab68

Style fix

48b0c75

Comment change

4f4a1e5

PratRanj07 requested a review from pranavrth December 27, 2024 11:30

		case RD_KAFKA_RESOURCE__CNT:
		return RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT;

		case RD_KAFKA_INTERNAL_RESOURCE_CONFIG_CNT:
		return RD_KAFKA_RESOURCE__CNT;

		rd_kafka_ConfigResourceType_t map_from_resource_type_to_config_resource_type(
		rd_kafka_ResourceType_t resourcetype) {

		rd_kafka_ResourceType_t map_from_config_resource_type_to_resource_type(
		rd_kafka_ConfigResourceType_t internal_resourcetype) {

	RD_KAFKA_CONFIG_RESOURCE_GROUP = 32, // Changed value for config APIs
	RD_KAFKA_CONFIG_RESOURCE_GROUP = 32,

		do_test_DescribeConfigs(rk, mainq);
		do_test_DescribeConfigs_groups(rk, mainq);

	const char *confs_set_append_group[] = {
	const char *confs_set_group[] = {

KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config #4939

Are you sure you want to change the base?

KIP 848:Extend DescribeConfigs and IncrementalAlterConfigs to support GROUP Config #4939

Conversation

PratRanj07 commented Dec 19, 2024

confluent-cla-assistant bot commented Dec 19, 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

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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment