[KIP-848] Tests for dev_kip848_fix_fast_subscribe_or_unsubscribe

Author: emasab

src/rd.h

@@ -226,6 +226,12 @@ static RD_INLINE RD_UNUSED char *rd_strndup(const char *s, size_t len) {
 #define RD_ARRAYSIZE(A)           RD_ARRAY_SIZE(A)
 #define RD_SIZEOF(TYPE, MEMBER)   sizeof(((TYPE *)NULL)->MEMBER)
 #define RD_OFFSETOF(TYPE, MEMBER) ((size_t) & (((TYPE *)NULL)->MEMBER))
+/** Array foreach */
+#define RD_ARRAY_FOREACH(ELEM, ARRAY, INDEX)                                   \
+        for ((INDEX = 0, (ELEM) = (ARRAY)[INDEX]);                             \
+             INDEX < RD_ARRAY_SIZE(ARRAY);                                     \
+             (ELEM) =                                                          \
+                 (++INDEX < RD_ARRAY_SIZE(ARRAY) ? (ARRAY)[INDEX] : (ELEM)))
 
 /**
  * Returns the 'I'th array element from static sized array 'A'

tests/0045-subscribe_update.c

@@ -819,6 +819,94 @@ static void do_test_resubscribe_with_regex() {
         SUB_TEST_PASS();
 }
 
+/**
+ * @brief Create many topics and apply several subscription
+ *        updates, unsubscribing and re-subscribing too.
+ *        After changing some subscriptions verifies that the assignment
+ *        corresponds to last one.
+ *
+ * @param with_rebalance_cb Use a rebalance callback to perform the assignment.
+ *                          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) {
+        char *topics[100] = {0};
+        char *topic;
+        size_t i;
+        char *group;
+        rd_kafka_t *rk;
+        rd_kafka_conf_t *conf;
+        const int subscription_size = 5, partition_cnt = 4;
+        rd_kafka_topic_partition_list_t *expected_assignment = NULL;
+
+        SUB_TEST("%s", with_rebalance_cb ? "with rebalance callback"
+                                         : "without rebalance callback");
+
+        RD_ARRAY_FOREACH(topic, topics, i) {
+                topics[i] = rd_strdup(test_mk_topic_name("topic", 1));
+        }
+        group = topics[0];
+
+        test_conf_init(&conf, NULL, 60);
+        if (with_rebalance_cb)
+                rd_kafka_conf_set_rebalance_cb(conf, test_rebalance_cb);
+        rk = test_create_consumer(group, NULL, conf, NULL);
+
+        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);
+        /* Give the cluster some more time to propagate metadata
+         * for 100 topics */
+        rd_sleep(1);
+
+        RD_ARRAY_FOREACH(topic, topics, i) {
+                int j, k;
+                rd_kafka_topic_partition_list_t *subscription =
+                    rd_kafka_topic_partition_list_new(subscription_size);
+                RD_IF_FREE(expected_assignment,
+                           rd_kafka_topic_partition_list_destroy);
+                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));
+                     j++) {
+                        topic = topics[j];
+                        rd_kafka_topic_partition_list_add(
+                            subscription, topic, RD_KAFKA_PARTITION_UA);
+                        /* Every 7 * 5 we unsubscribe and check that assignment
+                         * is empty. */
+                        if (i % (7 * 5) != 0)
+                                for (k = 0; k < partition_cnt; k++)
+                                        rd_kafka_topic_partition_list_add(
+                                            expected_assignment, topic, k);
+                }
+                TEST_CALL_ERR__(rd_kafka_subscribe(rk, subscription));
+                rd_kafka_topic_partition_list_destroy(subscription);
+
+                if (i % 7 == 0)
+                        TEST_CALL_ERR__(rd_kafka_unsubscribe(rk));
+                if (i % 5 == 0)
+                        test_consumer_wait_assignment_topic_partition_list(
+                            rk,
+                            /* poll when we have a rebalance callback */
+                            with_rebalance_cb, expected_assignment, 10000);
+        }
+        test_consumer_wait_assignment_topic_partition_list(
+            rk, with_rebalance_cb, expected_assignment, 10000);
+
+        TEST_CALL_ERR__(test_DeleteTopics_simple(rk, NULL, topics,
+                                                 RD_ARRAY_SIZE(topics), NULL));
+        RD_ARRAY_FOREACH(topic, topics, i) {
+                rd_free(topics[i]);
+        }
+
+        test_consumer_close(rk);
+        rd_kafka_destroy(rk);
+
+        SUB_TEST_PASS();
+}
+
 int main_0045_subscribe_update(int argc, char **argv) {
 
         if (!test_can_create_topics(1))
@@ -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) {
+        do_test_many_updates(rd_false);
+        do_test_many_updates(rd_true);
+        return 0;
+}
+
 int main_0045_subscribe_update_racks_mock(int argc, char **argv) {
         int use_replica_rack = 0;
         int use_client_rack  = 0;

tests/test.c

@@ -158,6 +158,7 @@ _TEST_DECL(0045_subscribe_update_non_exist_and_partchange);
 _TEST_DECL(0045_subscribe_update_mock);
 _TEST_DECL(0045_subscribe_update_racks_mock);
 _TEST_DECL(0045_resubscribe_with_regex);
+_TEST_DECL(0045_many_updates);
 _TEST_DECL(0046_rkt_cache);
 _TEST_DECL(0047_partial_buf_tmout);
 _TEST_DECL(0048_partitioner);
@@ -383,6 +384,7 @@ struct test tests[] = {
     _TEST(0045_subscribe_update_mock, TEST_F_LOCAL),
     _TEST(0045_subscribe_update_racks_mock, TEST_F_LOCAL),
     _TEST(0045_resubscribe_with_regex, 0, TEST_BRKVER(0, 9, 0, 0)),
+    _TEST(0045_many_updates, 0, TEST_BRKVER(0, 10, 2, 0)),
     _TEST(0046_rkt_cache, TEST_F_LOCAL),
     _TEST(0047_partial_buf_tmout, TEST_F_KNOWN_ISSUE),
     _TEST(0048_partitioner,
@@ -3093,6 +3095,99 @@ void test_consumer_verify_assignment0(const char *func,
         rd_kafka_topic_partition_list_destroy(assignment);
 }
 
+/**
+ * @brief Verify that the consumer's assignment matches the expected assignment.
+ *        passed as a topic partition list in \p expected_assignment .
+ */
+rd_bool_t test_consumer_verify_assignment_topic_partition_list0(
+    const char *func,
+    int line,
+    rd_kafka_t *rk,
+    rd_kafka_topic_partition_list_t *expected_assignment) {
+        rd_kafka_topic_partition_list_t *assignment;
+        rd_kafka_resp_err_t err;
+        int i;
+        rd_bool_t ret = rd_true;
+
+        if ((err = rd_kafka_assignment(rk, &assignment)))
+                TEST_FAIL("%s:%d: Failed to get assignment for %s: %s", func,
+                          line, rd_kafka_name(rk), rd_kafka_err2str(err));
+
+        TEST_SAY("%s assignment (%d partition(s)):\n", rd_kafka_name(rk),
+                 assignment->cnt);
+        for (i = 0; i < assignment->cnt; i++)
+                TEST_SAY(" %s [%" PRId32 "]\n", assignment->elems[i].topic,
+                         assignment->elems[i].partition);
+
+        rd_kafka_topic_partition_list_sort(assignment, NULL, NULL);
+        rd_kafka_topic_partition_list_sort(expected_assignment, NULL, NULL);
+
+        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;
+                        }
+                }
+        }
+
+done:
+        rd_kafka_topic_partition_list_destroy(assignment);
+        return ret;
+}
+
+/**
+ * @brief Wait until the consumer's assignment matches the expected assignment.
+ *        passed as a topic partition list in \p expected_assignment .
+ *        Polling if \p do_poll is true, otherwise sleeps.
+ *        Until \p timeout_ms milliseconds.
+ */
+void test_consumer_wait_assignment_topic_partition_list0(
+    const char *func,
+    int line,
+    rd_kafka_t *rk,
+    rd_bool_t do_poll,
+    rd_kafka_topic_partition_list_t *expected_assignment,
+    int timeout_ms) {
+        int i;
+        rd_ts_t end        = test_clock() + timeout_ms * 1000;
+        rd_bool_t verified = rd_false;
+
+        TEST_SAY("Verifying assignment\n");
+        rd_kafka_topic_partition_list_sort(expected_assignment, NULL, NULL);
+        TEST_SAY("%s expected assignment (%d partition(s)):\n",
+                 rd_kafka_name(rk), expected_assignment->cnt);
+        for (i = 0; i < expected_assignment->cnt; i++)
+                TEST_SAY(" %s [%" PRId32 "]\n",
+                         expected_assignment->elems[i].topic,
+                         expected_assignment->elems[i].partition);
+
+        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);
+
+        if (!verified) {
+                TEST_FAIL(
+                    "%s:%d: Expected assignment not found in %s's "
+                    "assignment",
+                    func, line, rd_kafka_name(rk));
+        }
+        TEST_SAY("Verified assignment\n");
+}
 
 
 /**

tests/test.h

@@ -676,6 +676,26 @@ void test_consumer_verify_assignment0(const char *func,
         test_consumer_verify_assignment0(__FUNCTION__, __LINE__, rk,           \
                                          fail_immediately, __VA_ARGS__)
 
+rd_bool_t test_consumer_verify_assignment_topic_partition_list0(
+    const char *func,
+    int line,
+    rd_kafka_t *rk,
+    rd_kafka_topic_partition_list_t *parts);
+#define test_consumer_verify_assignment_topic_partition_list(rk, parts)        \
+        test_consumer_verify_assignment_topic_partition_list0(                 \
+            __FUNCTION__, __LINE__, rk, parts)
+
+void test_consumer_wait_assignment_topic_partition_list0(
+    const char *func,
+    int line,
+    rd_kafka_t *rk,
+    rd_bool_t do_poll,
+    rd_kafka_topic_partition_list_t *expected_assignment,
+    int timeout_ms);
+#define test_consumer_wait_assignment_topic_partition_list(rk, do_poll, parts, \
+                                                           timeout_ms)         \
+        test_consumer_wait_assignment_topic_partition_list0(                   \
+            __FUNCTION__, __LINE__, rk, do_poll, parts, timeout_ms)
 
 void test_consumer_assign(const char *what,
                           rd_kafka_t *rk,