ListOffsets KIP : 396 by mahajanadhitya · Pull Request #4225 · confluentinc/librdkafka

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

… queue implemented

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

All committers have signed the CLA.

Emanuele Sabellico (emasab)

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

Emanuele Sabellico (emasab) · 2023-10-02T18:14:56Z

 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

Emanuele Sabellico (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.

rd_kafka_buf_set_maker and upgrade to version 7

rd_kafka_query_watermark_offsets if partition leader changes

Emanuele Sabellico (emasab) · 2023-10-12T17:56:09Z

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

Milind L (milindl) · 2023-10-13T04:02:07Z

+        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.

Milind L (milindl)

It mostly looks great, thanks for the PR mahajanadhitya and Emanuele Sabellico (@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.

Milind L (milindl) · 2023-10-13T09:44:06Z

+                                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

Milind L (milindl) · 2023-10-13T10:51:02Z

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


Move declaration up

Milind L (milindl) · 2023-10-13T10:53:11Z

+
+        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++)

Milind L (milindl) · 2023-10-13T10:53:20Z

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


Suggested change

}

Milind L (milindl) · 2023-10-13T11:17:33Z

+                        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

Milind L (milindl) · 2023-10-13T15:42:05Z

        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 */

Milind L (milindl) · 2023-10-13T15:42:16Z

        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 */

Milind L (milindl) · 2023-10-13T15:55:42Z

+
+        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

Milind L (milindl) · 2023-10-13T15:56:46Z

+
+                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

Emanuele Sabellico (emasab)

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

Milind L (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

Emanuele Sabellico (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

Emanuele Sabellico (emasab) requested changes Oct 2, 2023

View reviewed changes

Emanuele Sabellico (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

Milind L (milindl) reviewed Oct 13, 2023

View reviewed changes

Copyright

0f079d6

Milind L (milindl) reviewed Oct 13, 2023

View reviewed changes

Address comments

30b17ee

Emanuele Sabellico (emasab) approved these changes Oct 16, 2023

View reviewed changes

Merge branch 'master' into feature/listOffsets-AdminClient

052cbed

Milind L (milindl) approved these changes Oct 17, 2023

View reviewed changes

Update changelog with fixed

ce044a7

rd_kafka_query_watermark_offsets metadata refresh bug

Milind L (milindl) approved these changes Oct 17, 2023

View reviewed changes

Emanuele Sabellico (emasab) merged commit 99c67d3 into master Oct 17, 2023

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

Björn Svensson (bjosv) mentioned this pull request Oct 18, 2024

rd_kafka_query_watermark_offsets continues beyond timeout expiry #4878

Closed

7 tasks

	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 =

Conversation

mahajanadhitya commented Mar 17, 2023

Uh oh!

cla-assistant Bot commented Aug 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Emanuele Sabellico (emasab) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Emanuele Sabellico (emasab) Oct 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Emanuele Sabellico (emasab) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Emanuele Sabellico (emasab) commented Oct 12, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Milind L (milindl) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cla-assistant Bot commented Aug 21, 2023 •

edited

Loading

Emanuele Sabellico (emasab) Oct 2, 2023 •

edited

Loading