plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_assignor.c
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, "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);
}
/**
 * Maps the available topics to the group members' subscriptions
 * and updates the `member` map with the proper list of eligible topics,
 * the latter are returned in `eligible_topics`.
 */
Maps the available topics to the group members' subscriptions
and updates the `member` map with the proper list of eligible topics,
the latter are returned in `eligible_topics`.
Maps the available topics to the group members ' subscriptions and updates the ` member ` map with the proper list of eligible topics the latter are 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 (ti = 0 ; ti < metadata->topic_cnt ; ti++) {
                int i;
                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, "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 (i = 0 ; i < member_cnt ; i++) {
                        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);
}
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 ( ti = 0 ; ti < metadata -> topic_cnt ; ti ++ ) { int i ; 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 , " " , " " " \" \" " , 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 ( i = 0 ; i < member_cnt ; i ++ ) { 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 ) ; }
Maps the available topics to the group members' subscriptions
 and updates the `member` map with the proper list of eligible topics,
 the latter are returned in `eligible_topics`.
Maps the available topics to the group members ' subscriptions and updates the ` member ` map with the proper list of eligible topics the latter are returned in ` eligible_topics ` .
/* For each topic in the cluster, scan through the member list
         * to find matching subscriptions. */ /* Ignore topics in blacklist */ /* For each member: scan through its topic subscription */ /* Match topic against existing metadata,
                           incl regex matching. */