diff options
Diffstat (limited to 'fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_assignor.c')
| -rw-r--r-- | fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_assignor.c | 1065 |
1 files changed, 1065 insertions, 0 deletions
diff --git a/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_assignor.c b/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_assignor.c new file mode 100644 index 00000000..79257384 --- /dev/null +++ b/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_assignor.c @@ -0,0 +1,1065 @@ +/* + * librdkafka - The Apache Kafka C/C++ library + * + * Copyright (c) 2015 Magnus Edenhill + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ +#include "rdkafka_int.h" +#include "rdkafka_assignor.h" +#include "rdkafka_request.h" +#include "rdunittest.h" + +#include <ctype.h> + +/** + * Clear out and free any memory used by the member, but not the rkgm itself. + */ +void rd_kafka_group_member_clear(rd_kafka_group_member_t *rkgm) { + if (rkgm->rkgm_owned) + rd_kafka_topic_partition_list_destroy(rkgm->rkgm_owned); + + if (rkgm->rkgm_subscription) + rd_kafka_topic_partition_list_destroy(rkgm->rkgm_subscription); + + if (rkgm->rkgm_assignment) + rd_kafka_topic_partition_list_destroy(rkgm->rkgm_assignment); + + rd_list_destroy(&rkgm->rkgm_eligible); + + if (rkgm->rkgm_member_id) + rd_kafkap_str_destroy(rkgm->rkgm_member_id); + + if (rkgm->rkgm_group_instance_id) + rd_kafkap_str_destroy(rkgm->rkgm_group_instance_id); + + if (rkgm->rkgm_userdata) + rd_kafkap_bytes_destroy(rkgm->rkgm_userdata); + + if (rkgm->rkgm_member_metadata) + rd_kafkap_bytes_destroy(rkgm->rkgm_member_metadata); + + memset(rkgm, 0, sizeof(*rkgm)); +} + + +/** + * @brief Group member comparator (takes rd_kafka_group_member_t *) + */ +int rd_kafka_group_member_cmp(const void *_a, const void *_b) { + const rd_kafka_group_member_t *a = (const rd_kafka_group_member_t *)_a; + const rd_kafka_group_member_t *b = (const rd_kafka_group_member_t *)_b; + + /* Use the group instance id to compare static group members */ + if (!RD_KAFKAP_STR_IS_NULL(a->rkgm_group_instance_id) && + !RD_KAFKAP_STR_IS_NULL(b->rkgm_group_instance_id)) + return rd_kafkap_str_cmp(a->rkgm_group_instance_id, + b->rkgm_group_instance_id); + + return rd_kafkap_str_cmp(a->rkgm_member_id, b->rkgm_member_id); +} + + +/** + * Returns true if member subscribes to topic, else false. + */ +int rd_kafka_group_member_find_subscription(rd_kafka_t *rk, + const rd_kafka_group_member_t *rkgm, + const char *topic) { + int i; + + /* Match against member's subscription. */ + for (i = 0; i < rkgm->rkgm_subscription->cnt; i++) { + const rd_kafka_topic_partition_t *rktpar = + &rkgm->rkgm_subscription->elems[i]; + + if (rd_kafka_topic_partition_match(rk, rkgm, rktpar, topic, + NULL)) + return 1; + } + + return 0; +} + + +rd_kafkap_bytes_t *rd_kafka_consumer_protocol_member_metadata_new( + const rd_list_t *topics, + const void *userdata, + size_t userdata_size, + const rd_kafka_topic_partition_list_t *owned_partitions) { + + rd_kafka_buf_t *rkbuf; + rd_kafkap_bytes_t *kbytes; + int i; + int topic_cnt = rd_list_cnt(topics); + const rd_kafka_topic_info_t *tinfo; + size_t len; + + /* + * MemberMetadata => Version Subscription AssignmentStrategies + * Version => int16 + * Subscription => Topics UserData + * Topics => [String] + * UserData => Bytes + * OwnedPartitions => [Topic Partitions] // added in v1 + * Topic => string + * Partitions => [int32] + */ + + rkbuf = rd_kafka_buf_new(1, 100 + (topic_cnt * 100) + userdata_size); + + /* Version */ + rd_kafka_buf_write_i16(rkbuf, 1); + rd_kafka_buf_write_i32(rkbuf, topic_cnt); + RD_LIST_FOREACH(tinfo, topics, i) + rd_kafka_buf_write_str(rkbuf, tinfo->topic, -1); + if (userdata) + rd_kafka_buf_write_bytes(rkbuf, userdata, userdata_size); + else /* Kafka 0.9.0.0 can't parse NULL bytes, so we provide empty, + * which is compatible with all of the built-in Java client + * assignors at the present time (up to and including v2.5) */ + rd_kafka_buf_write_bytes(rkbuf, "", 0); + /* Following data is ignored by v0 consumers */ + if (!owned_partitions) + /* If there are no owned partitions, this is specified as an + * empty array, not NULL. */ + rd_kafka_buf_write_i32(rkbuf, 0); /* Topic count */ + else { + const rd_kafka_topic_partition_field_t fields[] = { + RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION, + RD_KAFKA_TOPIC_PARTITION_FIELD_END}; + rd_kafka_buf_write_topic_partitions( + rkbuf, owned_partitions, + rd_false /*don't skip invalid offsets*/, + rd_false /*any offset*/, fields); + } + + /* Get binary buffer and allocate a new Kafka Bytes with a copy. */ + rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf); + len = rd_slice_remains(&rkbuf->rkbuf_reader); + kbytes = rd_kafkap_bytes_new(NULL, (int32_t)len); + rd_slice_read(&rkbuf->rkbuf_reader, (void *)kbytes->data, len); + rd_kafka_buf_destroy(rkbuf); + + return kbytes; +} + + + +rd_kafkap_bytes_t *rd_kafka_assignor_get_metadata_with_empty_userdata( + const rd_kafka_assignor_t *rkas, + void *assignor_state, + const rd_list_t *topics, + const rd_kafka_topic_partition_list_t *owned_partitions) { + return rd_kafka_consumer_protocol_member_metadata_new(topics, NULL, 0, + owned_partitions); +} + + + +/** + * Returns 1 if all subscriptions are satifised for this member, else 0. + */ +static int rd_kafka_member_subscription_match( + rd_kafka_cgrp_t *rkcg, + rd_kafka_group_member_t *rkgm, + const rd_kafka_metadata_topic_t *topic_metadata, + rd_kafka_assignor_topic_t *eligible_topic) { + int i; + int has_regex = 0; + int matched = 0; + + /* Match against member's subscription. */ + for (i = 0; i < rkgm->rkgm_subscription->cnt; i++) { + const rd_kafka_topic_partition_t *rktpar = + &rkgm->rkgm_subscription->elems[i]; + int matched_by_regex = 0; + + if (rd_kafka_topic_partition_match(rkcg->rkcg_rk, rkgm, rktpar, + topic_metadata->topic, + &matched_by_regex)) { + rd_list_add(&rkgm->rkgm_eligible, + (void *)topic_metadata); + matched++; + has_regex += matched_by_regex; + } + } + + if (matched) + rd_list_add(&eligible_topic->members, rkgm); + + if (!has_regex && + rd_list_cnt(&rkgm->rkgm_eligible) == rkgm->rkgm_subscription->cnt) + return 1; /* All subscriptions matched */ + else + return 0; +} + + +static void rd_kafka_assignor_topic_destroy(rd_kafka_assignor_topic_t *at) { + rd_list_destroy(&at->members); + rd_free(at); +} + +int rd_kafka_assignor_topic_cmp(const void *_a, const void *_b) { + const rd_kafka_assignor_topic_t *a = + *(const rd_kafka_assignor_topic_t *const *)_a; + const rd_kafka_assignor_topic_t *b = + *(const rd_kafka_assignor_topic_t *const *)_b; + + return strcmp(a->metadata->topic, b->metadata->topic); +} + +/** + * Determine the complete set of topics that match at least one of + * the group member subscriptions. Associate with each of these the + * complete set of members that are subscribed to it. The result is + * returned in `eligible_topics`. + */ +static void +rd_kafka_member_subscriptions_map(rd_kafka_cgrp_t *rkcg, + rd_list_t *eligible_topics, + const rd_kafka_metadata_t *metadata, + rd_kafka_group_member_t *members, + int member_cnt) { + int ti; + rd_kafka_assignor_topic_t *eligible_topic = NULL; + + rd_list_init(eligible_topics, RD_MIN(metadata->topic_cnt, 10), + (void *)rd_kafka_assignor_topic_destroy); + + /* For each topic in the cluster, scan through the member list + * to find matching subscriptions. */ + for (ti = 0; ti < metadata->topic_cnt; ti++) { + int i; + + /* Ignore topics in blacklist */ + if (rkcg->rkcg_rk->rk_conf.topic_blacklist && + rd_kafka_pattern_match( + rkcg->rkcg_rk->rk_conf.topic_blacklist, + metadata->topics[ti].topic)) { + rd_kafka_dbg(rkcg->rkcg_rk, + TOPIC | RD_KAFKA_DBG_ASSIGNOR, "BLACKLIST", + "Assignor ignoring blacklisted " + "topic \"%s\"", + metadata->topics[ti].topic); + continue; + } + + if (!eligible_topic) + eligible_topic = rd_calloc(1, sizeof(*eligible_topic)); + + rd_list_init(&eligible_topic->members, member_cnt, NULL); + + /* For each member: scan through its topic subscription */ + for (i = 0; i < member_cnt; i++) { + /* Match topic against existing metadata, + incl regex matching. */ + rd_kafka_member_subscription_match( + rkcg, &members[i], &metadata->topics[ti], + eligible_topic); + } + + if (rd_list_empty(&eligible_topic->members)) { + rd_list_destroy(&eligible_topic->members); + continue; + } + + eligible_topic->metadata = &metadata->topics[ti]; + rd_list_add(eligible_topics, eligible_topic); + eligible_topic = NULL; + } + + if (eligible_topic) + rd_free(eligible_topic); +} + + +rd_kafka_resp_err_t rd_kafka_assignor_run(rd_kafka_cgrp_t *rkcg, + const rd_kafka_assignor_t *rkas, + rd_kafka_metadata_t *metadata, + rd_kafka_group_member_t *members, + int member_cnt, + char *errstr, + size_t errstr_size) { + rd_kafka_resp_err_t err; + rd_ts_t ts_start = rd_clock(); + int i; + rd_list_t eligible_topics; + int j; + + /* Construct eligible_topics, a map of: + * topic -> set of members that are subscribed to it. */ + rd_kafka_member_subscriptions_map(rkcg, &eligible_topics, metadata, + members, member_cnt); + + + if (rkcg->rkcg_rk->rk_conf.debug & + (RD_KAFKA_DBG_CGRP | RD_KAFKA_DBG_ASSIGNOR)) { + rd_kafka_dbg( + rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_ASSIGNOR, "ASSIGN", + "Group \"%s\" running %s assignor for " + "%d member(s) and " + "%d eligible subscribed topic(s):", + rkcg->rkcg_group_id->str, rkas->rkas_protocol_name->str, + member_cnt, eligible_topics.rl_cnt); + + for (i = 0; i < member_cnt; i++) { + const rd_kafka_group_member_t *member = &members[i]; + + rd_kafka_dbg( + rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_ASSIGNOR, + "ASSIGN", + " Member \"%.*s\"%s with " + "%d owned partition(s) and " + "%d subscribed topic(s):", + RD_KAFKAP_STR_PR(member->rkgm_member_id), + !rd_kafkap_str_cmp(member->rkgm_member_id, + rkcg->rkcg_member_id) + ? " (me)" + : "", + member->rkgm_owned ? member->rkgm_owned->cnt : 0, + member->rkgm_subscription->cnt); + for (j = 0; j < member->rkgm_subscription->cnt; j++) { + const rd_kafka_topic_partition_t *p = + &member->rkgm_subscription->elems[j]; + rd_kafka_dbg(rkcg->rkcg_rk, + CGRP | RD_KAFKA_DBG_ASSIGNOR, + "ASSIGN", " %s [%" PRId32 "]", + p->topic, p->partition); + } + } + } + + /* Call assignors assign callback */ + err = rkas->rkas_assign_cb( + rkcg->rkcg_rk, rkas, rkcg->rkcg_member_id->str, metadata, members, + member_cnt, (rd_kafka_assignor_topic_t **)eligible_topics.rl_elems, + eligible_topics.rl_cnt, errstr, errstr_size, rkas->rkas_opaque); + + if (err) { + rd_kafka_dbg( + rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_ASSIGNOR, "ASSIGN", + "Group \"%s\" %s assignment failed " + "for %d member(s): %s", + rkcg->rkcg_group_id->str, rkas->rkas_protocol_name->str, + (int)member_cnt, errstr); + } else if (rkcg->rkcg_rk->rk_conf.debug & + (RD_KAFKA_DBG_CGRP | RD_KAFKA_DBG_ASSIGNOR)) { + rd_kafka_dbg( + rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_ASSIGNOR, "ASSIGN", + "Group \"%s\" %s assignment for %d member(s) " + "finished in %.3fms:", + rkcg->rkcg_group_id->str, rkas->rkas_protocol_name->str, + (int)member_cnt, (float)(rd_clock() - ts_start) / 1000.0f); + for (i = 0; i < member_cnt; i++) { + const rd_kafka_group_member_t *member = &members[i]; + + rd_kafka_dbg(rkcg->rkcg_rk, + CGRP | RD_KAFKA_DBG_ASSIGNOR, "ASSIGN", + " Member \"%.*s\"%s assigned " + "%d partition(s):", + RD_KAFKAP_STR_PR(member->rkgm_member_id), + !rd_kafkap_str_cmp(member->rkgm_member_id, + rkcg->rkcg_member_id) + ? " (me)" + : "", + member->rkgm_assignment->cnt); + for (j = 0; j < member->rkgm_assignment->cnt; j++) { + const rd_kafka_topic_partition_t *p = + &member->rkgm_assignment->elems[j]; + rd_kafka_dbg(rkcg->rkcg_rk, + CGRP | RD_KAFKA_DBG_ASSIGNOR, + "ASSIGN", " %s [%" PRId32 "]", + p->topic, p->partition); + } + } + } + + rd_list_destroy(&eligible_topics); + + return err; +} + + +/** + * Assignor protocol string comparator + */ +static int rd_kafka_assignor_cmp_str(const void *_a, const void *_b) { + const char *a = _a; + const rd_kafka_assignor_t *b = _b; + + return rd_kafkap_str_cmp_str2(a, b->rkas_protocol_name); +} + +/** + * Find assignor by protocol name. + * + * Locality: any + * Locks: none + */ +rd_kafka_assignor_t *rd_kafka_assignor_find(rd_kafka_t *rk, + const char *protocol) { + return (rd_kafka_assignor_t *)rd_list_find( + &rk->rk_conf.partition_assignors, protocol, + rd_kafka_assignor_cmp_str); +} + + +/** + * Destroys an assignor (but does not unlink). + */ +static void rd_kafka_assignor_destroy(rd_kafka_assignor_t *rkas) { + rd_kafkap_str_destroy(rkas->rkas_protocol_type); + rd_kafkap_str_destroy(rkas->rkas_protocol_name); + rd_free(rkas); +} + + +/** + * @brief Check that the rebalance protocol of all enabled assignors is + * the same. + */ +rd_kafka_resp_err_t +rd_kafka_assignor_rebalance_protocol_check(const rd_kafka_conf_t *conf) { + int i; + rd_kafka_assignor_t *rkas; + rd_kafka_rebalance_protocol_t rebalance_protocol = + RD_KAFKA_REBALANCE_PROTOCOL_NONE; + + RD_LIST_FOREACH(rkas, &conf->partition_assignors, i) { + if (!rkas->rkas_enabled) + continue; + + if (rebalance_protocol == RD_KAFKA_REBALANCE_PROTOCOL_NONE) + rebalance_protocol = rkas->rkas_protocol; + else if (rebalance_protocol != rkas->rkas_protocol) + return RD_KAFKA_RESP_ERR__CONFLICT; + } + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/** + * @brief Add an assignor. + */ +rd_kafka_resp_err_t rd_kafka_assignor_add( + rd_kafka_t *rk, + const char *protocol_type, + const char *protocol_name, + rd_kafka_rebalance_protocol_t rebalance_protocol, + rd_kafka_resp_err_t (*assign_cb)( + rd_kafka_t *rk, + const struct rd_kafka_assignor_s *rkas, + const char *member_id, + const rd_kafka_metadata_t *metadata, + rd_kafka_group_member_t *members, + size_t member_cnt, + rd_kafka_assignor_topic_t **eligible_topics, + size_t eligible_topic_cnt, + char *errstr, + size_t errstr_size, + void *opaque), + rd_kafkap_bytes_t *(*get_metadata_cb)( + const struct rd_kafka_assignor_s *rkas, + void *assignor_state, + const rd_list_t *topics, + const rd_kafka_topic_partition_list_t *owned_partitions), + void (*on_assignment_cb)(const struct rd_kafka_assignor_s *rkas, + void **assignor_state, + const rd_kafka_topic_partition_list_t *assignment, + const rd_kafkap_bytes_t *userdata, + const rd_kafka_consumer_group_metadata_t *rkcgm), + void (*destroy_state_cb)(void *assignor_state), + int (*unittest_cb)(void), + void *opaque) { + rd_kafka_assignor_t *rkas; + + if (rd_kafkap_str_cmp_str(rk->rk_conf.group_protocol_type, + protocol_type)) + return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL; + + if (rebalance_protocol != RD_KAFKA_REBALANCE_PROTOCOL_COOPERATIVE && + rebalance_protocol != RD_KAFKA_REBALANCE_PROTOCOL_EAGER) + return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL; + + /* Dont overwrite application assignors */ + if ((rkas = rd_kafka_assignor_find(rk, protocol_name))) + return RD_KAFKA_RESP_ERR__CONFLICT; + + rkas = rd_calloc(1, sizeof(*rkas)); + + rkas->rkas_protocol_name = rd_kafkap_str_new(protocol_name, -1); + rkas->rkas_protocol_type = rd_kafkap_str_new(protocol_type, -1); + rkas->rkas_protocol = rebalance_protocol; + rkas->rkas_assign_cb = assign_cb; + rkas->rkas_get_metadata_cb = get_metadata_cb; + rkas->rkas_on_assignment_cb = on_assignment_cb; + rkas->rkas_destroy_state_cb = destroy_state_cb; + rkas->rkas_unittest = unittest_cb; + rkas->rkas_opaque = opaque; + rkas->rkas_index = INT_MAX; + + rd_list_add(&rk->rk_conf.partition_assignors, rkas); + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +/* Right trim string of whitespaces */ +static void rtrim(char *s) { + char *e = s + strlen(s); + + if (e == s) + return; + + while (e >= s && isspace(*e)) + e--; + + *e = '\0'; +} + + +static int rd_kafka_assignor_cmp_idx(const void *ptr1, const void *ptr2) { + const rd_kafka_assignor_t *rkas1 = (const rd_kafka_assignor_t *)ptr1; + const rd_kafka_assignor_t *rkas2 = (const rd_kafka_assignor_t *)ptr2; + return rkas1->rkas_index - rkas2->rkas_index; +} + + +/** + * Initialize assignor list based on configuration. + */ +int rd_kafka_assignors_init(rd_kafka_t *rk, char *errstr, size_t errstr_size) { + char *wanted; + char *s; + int idx = 0; + + rd_list_init(&rk->rk_conf.partition_assignors, 3, + (void *)rd_kafka_assignor_destroy); + + /* Initialize builtin assignors (ignore errors) */ + rd_kafka_range_assignor_init(rk); + rd_kafka_roundrobin_assignor_init(rk); + rd_kafka_sticky_assignor_init(rk); + + rd_strdupa(&wanted, rk->rk_conf.partition_assignment_strategy); + + s = wanted; + while (*s) { + rd_kafka_assignor_t *rkas = NULL; + char *t; + + /* Left trim */ + while (*s == ' ' || *s == ',') + s++; + + if ((t = strchr(s, ','))) { + *t = '\0'; + t++; + } else { + t = s + strlen(s); + } + + /* Right trim */ + rtrim(s); + + rkas = rd_kafka_assignor_find(rk, s); + if (!rkas) { + rd_snprintf(errstr, errstr_size, + "Unsupported partition.assignment.strategy:" + " %s", + s); + return -1; + } + + if (!rkas->rkas_enabled) { + rkas->rkas_enabled = 1; + rk->rk_conf.enabled_assignor_cnt++; + rkas->rkas_index = idx; + idx++; + } + + s = t; + } + + /* Sort the assignors according to the input strategy order + * since assignors will be scaned from the list sequentially + * and the strategies earlier in the list have higher priority. */ + rd_list_sort(&rk->rk_conf.partition_assignors, + rd_kafka_assignor_cmp_idx); + + /* Clear the SORTED flag because the list is sorted according to the + * rkas_index, but will do the search using rkas_protocol_name. */ + rk->rk_conf.partition_assignors.rl_flags &= ~RD_LIST_F_SORTED; + + if (rd_kafka_assignor_rebalance_protocol_check(&rk->rk_conf)) { + rd_snprintf(errstr, errstr_size, + "All partition.assignment.strategy (%s) assignors " + "must have the same protocol type, " + "online migration between assignors with " + "different protocol types is not supported", + rk->rk_conf.partition_assignment_strategy); + return -1; + } + + return 0; +} + + + +/** + * Free assignors + */ +void rd_kafka_assignors_term(rd_kafka_t *rk) { + rd_list_destroy(&rk->rk_conf.partition_assignors); +} + + + +/** + * @brief Unittest for assignors + */ +static int ut_assignors(void) { + const struct { + const char *name; + int topic_cnt; + struct { + const char *name; + int partition_cnt; + } topics[12]; + int member_cnt; + struct { + const char *name; + int topic_cnt; + const char *topics[12]; + } members[3]; + int expect_cnt; + struct { + const char *protocol_name; + struct { + int partition_cnt; + const char *partitions[12]; /* "topic:part" */ + } members[3]; + } expect[2]; + } tests[] = { + /* + * Test cases + */ + { + .name = "Symmetrical subscription", + .topic_cnt = 4, + .topics = + { + {"a", 3}, /* a:0 a:1 a:2 */ + { + "b", + 4, + }, /* b:0 b:1 b:2 b:3 */ + {"c", 2}, /* c:0 c:1 */ + {"d", 1}, /* d:0 */ + }, + .member_cnt = 2, + .members = + { + {.name = "consumer1", + .topic_cnt = 4, + .topics = {"d", "b", "a", "c"}}, + {.name = "consumer2", + .topic_cnt = 4, + .topics = {"a", "b", "c", "d"}}, + }, + .expect_cnt = 2, + .expect = + { + { + .protocol_name = "range", + .members = + { + /* Consumer1 */ + {6, + {"a:0", "a:1", "b:0", "b:1", "c:0", + "d:0"}}, + /* Consumer2 */ + {4, {"a:2", "b:2", "b:3", "c:1"}}, + }, + }, + { + .protocol_name = "roundrobin", + .members = + { + /* Consumer1 */ + {5, {"a:0", "a:2", "b:1", "b:3", "c:1"}}, + /* Consumer2 */ + {5, {"a:1", "b:0", "b:2", "c:0", "d:0"}}, + }, + }, + }, + }, + { + .name = "1*3 partitions (asymmetrical)", + .topic_cnt = 1, + .topics = + { + {"a", 3}, + }, + .member_cnt = 2, + .members = + { + {.name = "consumer1", + .topic_cnt = 3, + .topics = {"a", "b", "c"}}, + {.name = "consumer2", .topic_cnt = 1, .topics = {"a"}}, + }, + .expect_cnt = 2, + .expect = + { + { + .protocol_name = "range", + .members = + { + /* Consumer1. + * range assignor applies + * per topic. */ + {2, {"a:0", "a:1"}}, + /* Consumer2 */ + {1, {"a:2"}}, + }, + }, + { + .protocol_name = "roundrobin", + .members = + { + /* Consumer1 */ + {2, {"a:0", "a:2"}}, + /* Consumer2 */ + {1, {"a:1"}}, + }, + }, + }, + }, + { + .name = "#2121 (asymmetrical)", + .topic_cnt = 12, + .topics = + { + {"a", 1}, + {"b", 1}, + {"c", 1}, + {"d", 1}, + {"e", 1}, + {"f", 1}, + {"g", 1}, + {"h", 1}, + {"i", 1}, + {"j", 1}, + {"k", 1}, + {"l", 1}, + }, + .member_cnt = 2, + .members = + { + { + .name = "consumer1", + .topic_cnt = 12, + .topics = + { + "a", + "b", + "c", + "d", + "e", + "f", + "g", + "h", + "i", + "j", + "k", + "l", + }, + }, + { + .name = "consumer2", /* must be second */ + .topic_cnt = 5, + .topics = + { + "b", + "d", + "f", + "h", + "l", + }, + }, + }, + .expect_cnt = 2, + .expect = + { + { + .protocol_name = "range", + .members = + { + /* Consumer1. + * All partitions. */ + {12, + { + "a:0", + "b:0", + "c:0", + "d:0", + "e:0", + "f:0", + "g:0", + "h:0", + "i:0", + "j:0", + "k:0", + "l:0", + }}, + /* Consumer2 */ + {0}, + }, + }, + { + .protocol_name = "roundrobin", + .members = + { + /* Consumer1 */ + { + 7, + { + "a:0", + "c:0", + "e:0", + "g:0", + "i:0", + "j:0", + "k:0", + }, + }, + /* Consumer2 */ + {5, {"b:0", "d:0", "f:0", "h:0", "l:0"}}, + }, + }, + }, + }, + {NULL}, + }; + rd_kafka_conf_t *conf; + rd_kafka_t *rk; + const rd_kafka_assignor_t *rkas; + int fails = 0; + int i; + + conf = rd_kafka_conf_new(); + rd_kafka_conf_set(conf, "group.id", "group", NULL, 0); + rd_kafka_conf_set(conf, "debug", rd_getenv("TEST_DEBUG", NULL), NULL, + 0); + rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0); + RD_UT_ASSERT(rk != NULL, "Failed to create consumer"); + + /* Run through test cases */ + for (i = 0; tests[i].name; i++) { + int ie, it, im; + rd_kafka_metadata_t metadata; + rd_kafka_group_member_t *members; + + /* Create topic metadata */ + metadata.topic_cnt = tests[i].topic_cnt; + metadata.topics = + rd_alloca(sizeof(*metadata.topics) * metadata.topic_cnt); + memset(metadata.topics, 0, + sizeof(*metadata.topics) * metadata.topic_cnt); + for (it = 0; it < metadata.topic_cnt; it++) { + metadata.topics[it].topic = + (char *)tests[i].topics[it].name; + metadata.topics[it].partition_cnt = + tests[i].topics[it].partition_cnt; + metadata.topics[it].partitions = NULL; /* Not used */ + } + + /* Create members */ + members = rd_alloca(sizeof(*members) * tests[i].member_cnt); + memset(members, 0, sizeof(*members) * tests[i].member_cnt); + + for (im = 0; im < tests[i].member_cnt; im++) { + rd_kafka_group_member_t *rkgm = &members[im]; + rkgm->rkgm_member_id = + rd_kafkap_str_new(tests[i].members[im].name, -1); + rkgm->rkgm_group_instance_id = + rd_kafkap_str_new(tests[i].members[im].name, -1); + rd_list_init(&rkgm->rkgm_eligible, + tests[i].members[im].topic_cnt, NULL); + + rkgm->rkgm_subscription = + rd_kafka_topic_partition_list_new( + tests[i].members[im].topic_cnt); + for (it = 0; it < tests[i].members[im].topic_cnt; it++) + rd_kafka_topic_partition_list_add( + rkgm->rkgm_subscription, + tests[i].members[im].topics[it], + RD_KAFKA_PARTITION_UA); + + rkgm->rkgm_userdata = NULL; + + rkgm->rkgm_assignment = + rd_kafka_topic_partition_list_new( + rkgm->rkgm_subscription->size); + } + + /* For each assignor verify that the assignment + * matches the expection set out in the test case. */ + for (ie = 0; ie < tests[i].expect_cnt; ie++) { + rd_kafka_resp_err_t err; + char errstr[256]; + + RD_UT_SAY("Test case %s: %s assignor", tests[i].name, + tests[i].expect[ie].protocol_name); + + if (!(rkas = rd_kafka_assignor_find( + rk, tests[i].expect[ie].protocol_name))) { + RD_UT_FAIL( + "Assignor test case %s for %s failed: " + "assignor not found", + tests[i].name, + tests[i].expect[ie].protocol_name); + } + + /* Run assignor */ + err = rd_kafka_assignor_run( + rk->rk_cgrp, rkas, &metadata, members, + tests[i].member_cnt, errstr, sizeof(errstr)); + + RD_UT_ASSERT(!err, "Assignor case %s for %s failed: %s", + tests[i].name, + tests[i].expect[ie].protocol_name, errstr); + + /* Verify assignments */ + for (im = 0; im < tests[i].member_cnt; im++) { + rd_kafka_group_member_t *rkgm = &members[im]; + int ia; + + if (rkgm->rkgm_assignment->cnt != + tests[i] + .expect[ie] + .members[im] + .partition_cnt) { + RD_UT_WARN( + " Member %.*s assignment count " + "mismatch: %d != %d", + RD_KAFKAP_STR_PR( + rkgm->rkgm_member_id), + rkgm->rkgm_assignment->cnt, + tests[i] + .expect[ie] + .members[im] + .partition_cnt); + fails++; + } + + if (rkgm->rkgm_assignment->cnt > 0) + rd_kafka_topic_partition_list_sort_by_topic( + rkgm->rkgm_assignment); + + for (ia = 0; ia < rkgm->rkgm_assignment->cnt; + ia++) { + rd_kafka_topic_partition_t *p = + &rkgm->rkgm_assignment->elems[ia]; + char part[64]; + const char *exp = + ia < tests[i] + .expect[ie] + .members[im] + .partition_cnt + ? tests[i] + .expect[ie] + .members[im] + .partitions[ia] + : "(none)"; + + rd_snprintf(part, sizeof(part), "%s:%d", + p->topic, + (int)p->partition); + +#if 0 /* Enable to print actual assignment */ + RD_UT_SAY(" Member %.*s assignment " + "%d/%d %s =? %s", + RD_KAFKAP_STR_PR( + rkgm->rkgm_member_id), + ia, + rkgm->rkgm_assignment->cnt-1, + part, exp); +#endif + + if (strcmp(part, exp)) { + RD_UT_WARN( + " Member %.*s " + "assignment %d/%d " + "mismatch: %s != %s", + RD_KAFKAP_STR_PR( + rkgm->rkgm_member_id), + ia, + rkgm->rkgm_assignment->cnt - + 1, + part, exp); + fails++; + } + } + + /* Reset assignment for next loop */ + rd_kafka_topic_partition_list_destroy( + rkgm->rkgm_assignment); + rkgm->rkgm_assignment = + rd_kafka_topic_partition_list_new( + rkgm->rkgm_subscription->size); + } + } + + for (im = 0; im < tests[i].member_cnt; im++) { + rd_kafka_group_member_t *rkgm = &members[im]; + rd_kafka_group_member_clear(rkgm); + } + } + + + /* Run assignor-specific unittests */ + RD_LIST_FOREACH(rkas, &rk->rk_conf.partition_assignors, i) { + if (rkas->rkas_unittest) + fails += rkas->rkas_unittest(); + } + + rd_kafka_destroy(rk); + + if (fails) + return 1; + + RD_UT_PASS(); +} + + +/** + * @brief Unit tests for assignors + */ +int unittest_assignors(void) { + return ut_assignors(); +} |