ListOffsets KIP : 396 #4225

mahajanadhitya · 2023-03-17T18:37:33Z

… queue implemented

cla-assistant · 2023-08-21T15:00:40Z

All committers have signed the CLA.

emasab

First review comments about general checks and rdkafka.h additions.

examples/list_offsets.c

src/rdkafka.h

examples/list_offsets.c

src/rdkafka.h

examples/list_offsets

src/rdkafka.h

emasab · 2023-10-02T18:14:56Z

src/rdkafka.h

@@ -9732,6 +9767,26 @@ rd_kafka_error_t *rd_kafka_commit_transaction(rd_kafka_t *rk, int timeout_ms);
 RD_EXPORT
 rd_kafka_error_t *rd_kafka_abort_transaction(rd_kafka_t *rk, int timeout_ms);

+typedef enum rd_kafka_OffsetSpec_s {


Missing to move the section.
You can also put it after DeleteConsumerGroupOffsets, before Admin API - User SCRAM credentials

src/rdkafka_admin.h

src/rdkafka.h

emasab

There are some things to change to keep backward compatibility for consumer rd_kafka_query_watermark_offsets and rd_kafka_offsets_for_times functions and to upgrade internal consumer calls too.

src/rdkafka_admin.c

src/rdkafka_admin.h

src/rdkafka_request.c

rd_kafka_buf_set_maker and upgrade to version 7

rd_kafka_query_watermark_offsets if partition leader changes

emasab · 2023-10-12T17:56:09Z

rd_kafka_offsets_for_times needs to be fixed as well but will be in a different PR

milindl · 2023-10-13T04:02:07Z

src/rdkafka.c

+        if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
+                /* Remove its cache in case the topic isn't a known topic. */
+                rd_kafka_wrlock(rk);
+                rd_kafka_metadata_cache_delete_by_name(rk, state->topic);


I think we can check here also that if cache is hinted for this topic, and in that case, do nothing as a refresh is already ongoing.

Same for ListOffsets

This is already ensured by rd_kafka_metadata_cache_delete_by_name because it does a rd_kafka_metadata_cache_find with valid equal to 1.

milindl

It mostly looks great, thanks for the PR @mahajanadhitya and @emasab !
Some minor comments about the conventions and declarations of the variables
Two comments for the tests.
Good fix for the not leader issue, let's slowly (in other PRs) see if we can extend it to other admin functions which have similar retriable errors.

CHANGELOG.md

milindl · 2023-10-13T09:44:06Z

tests/0081-admin.c

+                                rd_usleep(100000, 0);
+                        } else if (err) {
+                                TEST_FAIL("Failed with error: %s",
+                                          rd_kafka_err2name(err));


nit: break the outer loop here, it's cleaner that way without the else

milindl · 2023-10-13T10:51:02Z

src/rdkafka_admin.c

+                goto err;
+        }
+
+        rd_list_t *topic_partitions_sorted = rd_list_new(


Move declaration up

milindl · 2023-10-13T10:53:11Z

src/rdkafka_admin.c

+
+        rd_list_t *topic_partitions_sorted = rd_list_new(
+            topic_partitions->cnt, rd_kafka_topic_partition_destroy_free);
+        for (i = 0; i < topic_partitions->cnt; i++) {


Suggested change

for (i = 0; i < topic_partitions->cnt; i++) {

for (i = 0; i < topic_partitions->cnt; i++)

milindl · 2023-10-13T10:53:20Z

src/rdkafka_admin.c

+                rd_list_add(
+                    topic_partitions_sorted,
+                    rd_kafka_topic_partition_copy(&topic_partitions->elems[i]));
+        }


Suggested change

}

milindl · 2023-10-13T11:17:33Z

tests/0081-admin.c

+                        err = rd_kafka_event_error(event);
+                        if (err == RD_KAFKA_RESP_ERR__NOENT) {
+                                /* Still looking for the leader */
+                                rd_usleep(100000, 0);


nit: continue the loop here, avoids if elses

milindl · 2023-10-13T15:42:05Z

src/rdkafka.h

@@ -6829,7 +6850,9 @@ typedef enum rd_kafka_admin_op_t {
        RD_KAFKA_ADMIN_OP_ALTERUSERSCRAMCREDENTIALS,
        RD_KAFKA_ADMIN_OP_DESCRIBETOPICS,  /**< DescribeTopics */
        RD_KAFKA_ADMIN_OP_DESCRIBECLUSTER, /**< DescribeCluster */
-        RD_KAFKA_ADMIN_OP__CNT             /**< Number of ops defined */
+        /** ListOffsets */
+        RD_KAFKA_ADMIN_OP_LISTOFFSETS,


Suggested change

RD_KAFKA_ADMIN_OP_LISTOFFSETS,

RD_KAFKA_ADMIN_OP_LISTOFFSETS, /**< ListOffsets */

milindl · 2023-10-13T15:42:16Z

src/rdkafka.h

@@ -6829,7 +6850,9 @@ typedef enum rd_kafka_admin_op_t {
        RD_KAFKA_ADMIN_OP_ALTERUSERSCRAMCREDENTIALS,
        RD_KAFKA_ADMIN_OP_DESCRIBETOPICS,  /**< DescribeTopics */
        RD_KAFKA_ADMIN_OP_DESCRIBECLUSTER, /**< DescribeCluster */
-        RD_KAFKA_ADMIN_OP__CNT             /**< Number of ops defined */
+        /** ListOffsets */


Suggested change

/** ListOffsets */

milindl · 2023-10-13T15:55:42Z

tests/0138-admin_mock.c

+
+        rd_kafka_event_destroy(rkev);
+
+        rd_kafka_mock_broker_push_request_error_rtts(


Similar to the watermarks test - replace push of errors with rd_kafka_mock_partition_set_leader(mcluster, topic, 0, 2); instead and let the mock handler introduce the error

milindl · 2023-10-13T15:56:46Z

tests/0081-admin.c

+
+                TEST_SAY("Testing offset %" PRId64 "\n", test_fixture.query);
+
+                rd_kafka_topic_partition_list_t *topic_partitions_copy =


declaration might need to be moved up

emasab

I approve. Given there are my changes too, let's wait for a second approval by @milindl

milindl

Thanks for these changes!

rd_kafka_query_watermark_offsets metadata refresh bug

mahajanadhitya force-pushed the feature/listOffsets-AdminClient branch from cbd499c to 9f89db5 Compare May 26, 2023 11:42

mahajanadhitya force-pushed the feature/listOffsets-AdminClient branch from 6b35ec6 to 3689716 Compare July 18, 2023 09:58

emasab requested changes Sep 4, 2023

View reviewed changes

mahajanadhitya force-pushed the feature/listOffsets-AdminClient branch 3 times, most recently from 09524c4 to 6fcce61 Compare September 28, 2023 06:16

emasab requested changes Oct 2, 2023

View reviewed changes

emasab requested changes Oct 3, 2023

View reviewed changes

mahajanadhitya and others added 13 commits October 11, 2023 14:50

Rebase Commit

3a6c7dc

Changes

fe1a058

PR Addressal

a8b3759

Minor

690546d

Minor

4405faf

Style fix

a4bcc02

Remove compiled file

379233a

Remove new include

4f9a99e

Avoid regression by preserving

3fd524e

rd_kafka_buf_set_maker and upgrade to version 7

Improve documentation

1e88fd1

Change doxygen groups order

9f3de7f

Test refactor

fcc4a3d

Test earlier versions

f97979d

mahajanadhitya force-pushed the feature/listOffsets-AdminClient branch from 8f43390 to f97979d Compare October 11, 2023 09:20

mahajanadhitya and others added 5 commits October 11, 2023 14:57

Warnings

f6f00a1

Changelog and introduction

54d4752

Upgrade Mock handler

b76ed22

Failing tests

ae4cdaa

Clear metadata cache for ListOffsets or

4e431d0

rd_kafka_query_watermark_offsets if partition leader changes

milindl reviewed Oct 13, 2023

View reviewed changes

Copyright

0f079d6

milindl reviewed Oct 13, 2023

View reviewed changes

Address comments

30b17ee

emasab approved these changes Oct 16, 2023

View reviewed changes

Merge branch 'master' into feature/listOffsets-AdminClient

052cbed

milindl approved these changes Oct 17, 2023

View reviewed changes

Update changelog with fixed

ce044a7

rd_kafka_query_watermark_offsets metadata refresh bug

milindl approved these changes Oct 17, 2023

View reviewed changes

emasab merged commit 99c67d3 into master Oct 17, 2023
1 check was pending

emasab deleted the feature/listOffsets-AdminClient branch October 17, 2023 08:59

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ListOffsets KIP : 396 #4225

ListOffsets KIP : 396 #4225

mahajanadhitya commented Mar 17, 2023

cla-assistant bot commented Aug 21, 2023 •

edited

Loading

emasab left a comment

emasab Oct 2, 2023 •

edited

Loading

emasab left a comment

emasab commented Oct 12, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

emasab Oct 13, 2023

milindl left a comment

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

milindl Oct 13, 2023

emasab left a comment

milindl left a comment

	for (i = 0; i < topic_partitions->cnt; i++) {
	for (i = 0; i < topic_partitions->cnt; i++)

	RD_KAFKA_ADMIN_OP_LISTOFFSETS,
	RD_KAFKA_ADMIN_OP_LISTOFFSETS, /*< ListOffsets /


		rd_kafka_event_destroy(rkev);

		rd_kafka_mock_broker_push_request_error_rtts(


		TEST_SAY("Testing offset %" PRId64 "\n", test_fixture.query);

		rd_kafka_topic_partition_list_t *topic_partitions_copy =

ListOffsets KIP : 396 #4225

ListOffsets KIP : 396 #4225

Conversation

mahajanadhitya commented Mar 17, 2023

cla-assistant bot commented Aug 21, 2023 • edited Loading

emasab left a comment

Choose a reason for hiding this comment

emasab Oct 2, 2023 • edited Loading

Choose a reason for hiding this comment

emasab left a comment

Choose a reason for hiding this comment

emasab commented Oct 12, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

milindl 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

emasab left a comment

Choose a reason for hiding this comment

milindl left a comment

Choose a reason for hiding this comment

cla-assistant bot commented Aug 21, 2023 •

edited

Loading

emasab Oct 2, 2023 •

edited

Loading