[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe #5026

emasab · 2025-04-08T18:21:58Z

No description provided.

confluent-cla-assistant · 2025-04-08T18:22:08Z

🎉 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 · 2025-04-09T09:51:21Z

tests/0045-subscribe_update.c

+        TEST_SAY("Creating %" PRIusz " topics\n", RD_ARRAY_SIZE(topics));
+        TEST_CALL_ERR__(test_CreateTopics_simple(
+            rk, NULL, topics, RD_ARRAY_SIZE(topics), partition_cnt, NULL));
+        test_wait_topic_exists(rk, topics[0], 5000);


Better to check for last topic.

pranavrth · 2025-04-09T11:10:51Z

tests/0045-subscribe_update.c

+                RD_IF_FREE(expected_assignment,
+                           rd_kafka_topic_partition_list_destroy);
+                expected_assignment = rd_kafka_topic_partition_list_new(
+                    subscription_size * partition_cnt);


subscription_size * partition_cnt is constant throughout the loop. Extract it in a variable.

pranavrth · 2025-04-09T11:13:18Z

tests/0045-subscribe_update.c

+                expected_assignment = rd_kafka_topic_partition_list_new(
+                    subscription_size * partition_cnt);
+                for (j = i; j < (int)RD_MIN(i + subscription_size,
+                                            RD_ARRAY_SIZE(topics));


No need to use RD_ARRAY_SIZE(topics)). Directly extract the size in a define and use it even when initializing the array.

pranavrth · 2025-04-09T13:29:22Z

tests/0045-subscribe_update.c

+                        topic = topics[j];
+                        rd_kafka_topic_partition_list_add(
+                            subscription, topic, RD_KAFKA_PARTITION_UA);
+                        /* Every 7 * 5 we unsubscribe and check that assignment


Suggested change

/* Every 7 * 5 we unsubscribe and check that assignment

/* We unsubscribe every 7 iteration and we check assignments every 5 iteration so at 7 * 5 unsubscribe and check that assignment

pranavrth · 2025-04-09T13:32:15Z

tests/0045-subscribe_update.c

+        char *group;
+        rd_kafka_t *rk;
+        rd_kafka_conf_t *conf;
+        const int subscription_size = 5, partition_cnt = 4;


Suggested change

const int subscription_size = 5, partition_cnt = 4;

const int max_subscription_size = 5, partition_cnt = 4;

pranavrth · 2025-04-09T13:50:16Z

tests/test.c

+        do {
+                verified =
+                    test_consumer_verify_assignment_topic_partition_list0(
+                        func, line, rk, expected_assignment);
+                if (!verified) {
+                        if (do_poll)
+                                test_consumer_poll_once(rk, NULL, 100);
+                        else
+                                rd_usleep(100 * 1000, NULL);
+                }
+        } while (test_clock() < end && !verified);


No need for do-while loop. Normal while will work. We can remove checking !verified at 2 places.

Suggested change

do {

verified =

test_consumer_verify_assignment_topic_partition_list0(

func, line, rk, expected_assignment);

if (!verified) {

if (do_poll)

test_consumer_poll_once(rk, NULL, 100);

else

rd_usleep(100 * 1000, NULL);

}

} while (test_clock() < end && !verified);

while (test_clock() < end) {

verified =

test_consumer_verify_assignment_topic_partition_list0(

func, line, rk, expected_assignment);

if(verified)

break;

if (do_poll)

test_consumer_poll_once(rk, NULL, 100);

else

rd_usleep(100 * 1000, NULL);

}

That way we still need to call

verified = test_consumer_verify_assignment_topic_partition_list0( func, line, rk, expected_assignment);

before of after the loop in case we call it with 0 timeout, I use the break but keep the do ... while

pranavrth · 2025-04-09T13:51:11Z

tests/test.c

+        } while (test_clock() < end && !verified);
+
+        if (!verified) {
+                TEST_FAIL(


return after test fail.

not needed, TEST_FAIL either calls thrd_exit or assert

pranavrth · 2025-04-09T13:53:34Z

tests/test.c

+        rd_bool_t verified = rd_false;
+
+        TEST_SAY("Verifying assignment\n");
+        rd_kafka_topic_partition_list_sort(expected_assignment, NULL, NULL);


We shouldn't sort the list here. Its the responsibility of the test_consumer_verify_assignment_topic_partition_list0 on how to verify.

pranavrth · 2025-04-09T13:55:23Z

tests/test.c

+        rd_kafka_topic_partition_list_sort(assignment, NULL, NULL);
+        rd_kafka_topic_partition_list_sort(expected_assignment, NULL, NULL);


Ideally we should not be sorting these lists. Caller will expect it to be as it is. I would prefer creating new list for expected_assignment, match that and then destroy that. assignment is local so we can do anything with it.

pranavrth · 2025-04-09T14:03:02Z

tests/test.c

+        if (assignment->cnt != expected_assignment->cnt) {
+                ret = rd_false;
+                goto done;
+        } else {
+                for (i = 0; i < assignment->cnt; i++) {
+                        if (strcmp(assignment->elems[i].topic,
+                                   expected_assignment->elems[i].topic) ||
+                            assignment->elems[i].partition !=
+                                expected_assignment->elems[i].partition) {
+                                ret = rd_false;
+                                goto done;
+                        }
+                }
+        }


goto done or break. Whatever you want.

Suggested change

if (assignment->cnt != expected_assignment->cnt) {

ret = rd_false;

goto done;

} else {

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

if (strcmp(assignment->elems[i].topic,

expected_assignment->elems[i].topic) ||

assignment->elems[i].partition !=

expected_assignment->elems[i].partition) {

ret = rd_false;

goto done;

}

}

}

if (assignment->cnt != expected_assignment->cnt) {

ret = rd_false;

goto done;

}

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

if (strcmp(assignment->elems[i].topic,

expected_assignment->elems[i].topic) ||

assignment->elems[i].partition !=

expected_assignment->elems[i].partition) {

ret = rd_false;

break;

}

}

pranavrth · 2025-04-09T14:07:16Z

tests/0045-subscribe_update.c

@@ -862,6 +950,12 @@ int main_0045_resubscribe_with_regex(int argc, char **argv) {
        return 0;
 }

+int main_0045_many_updates(int argc, char **argv) {


Suggested change

int main_0045_many_updates(int argc, char **argv) {

int main_0045_subscribe_many_updates(int argc, char **argv) {

pranavrth · 2025-04-09T14:07:52Z

tests/0045-subscribe_update.c

+ *                          It needs to poll the consumer when awaiting for the
+ *                          assignment in this case.
+ */
+static void do_test_many_updates(rd_bool_t with_rebalance_cb) {


Suggested change

static void do_test_many_updates(rd_bool_t with_rebalance_cb) {

static void do_test_subscribe_many_updates(rd_bool_t with_rebalance_cb) {

emasab requested a review from a team as a code owner April 8, 2025 18:21

Base automatically changed from dev_kip848_fix_fast_subscribe_or_unsubscribe to master April 8, 2025 19:09

airlock-confluentinc bot force-pushed the dev_kip848_fix_fast_subscribe_or_unsubscribe_tests branch from 0999c46 to 050b06c Compare April 9, 2025 09:37

pranavrth requested changes Apr 9, 2025

View reviewed changes

emasab mentioned this pull request Jun 26, 2025

[KIP-848] Tests for: Fix for a rapid unsubscribe while the member id is still not assigned #4700

Merged

emasab added 2 commits June 26, 2025 12:08

[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe

4210c10

Address comments

93df6bf

airlock-confluentinc bot force-pushed the dev_kip848_fix_fast_subscribe_or_unsubscribe_tests branch from 050b06c to 93df6bf Compare June 26, 2025 10:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe #5026

[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe #5026

Uh oh!

emasab commented Apr 8, 2025

Uh oh!

confluent-cla-assistant bot commented Apr 8, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

emasab Jun 26, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

emasab Jun 26, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

pranavrth Apr 9, 2025

Uh oh!

Uh oh!

	/* Every 7 * 5 we unsubscribe and check that assignment
	/* We unsubscribe every 7 iteration and we check assignments every 5 iteration so at 7 * 5 unsubscribe and check that assignment

	const int subscription_size = 5, partition_cnt = 4;
	const int max_subscription_size = 5, partition_cnt = 4;

		rd_kafka_topic_partition_list_sort(assignment, NULL, NULL);
		rd_kafka_topic_partition_list_sort(expected_assignment, NULL, NULL);

	int main_0045_many_updates(int argc, char **argv) {
	int main_0045_subscribe_many_updates(int argc, char **argv) {

	static void do_test_many_updates(rd_bool_t with_rebalance_cb) {
	static void do_test_subscribe_many_updates(rd_bool_t with_rebalance_cb) {

[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe #5026

Are you sure you want to change the base?

[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe #5026

Uh oh!

Conversation

emasab commented Apr 8, 2025

Uh oh!

confluent-cla-assistant bot commented Apr 8, 2025

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!

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!

Uh oh!