plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
void rd_kafka_cgrp_metadata_update_check (rd_kafka_cgrp_t *rkcg, int do_join) {
        rd_list_t *tinfos;
        rd_kafka_topic_partition_list_t *errored;

        rd_kafka_assert(NULL, thrd_is_current(rkcg->rkcg_rk->rk_thread));

        if (!rkcg->rkcg_subscription || rkcg->rkcg_subscription->cnt == 0)
                return;

        /*
         * Unmatched topics will be added to the errored list.
         */
        errored = rd_kafka_topic_partition_list_new(0);

        /*
         * Create a list of the topics in metadata that matches our subscription
         */
        tinfos = rd_list_new(rkcg->rkcg_subscription->cnt,
                             (void *)rd_kafka_topic_info_destroy);

        if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION)
                rd_kafka_metadata_topic_match(rkcg->rkcg_rk,
                                              tinfos, rkcg->rkcg_subscription,
                                              errored);
        else
                rd_kafka_metadata_topic_filter(rkcg->rkcg_rk,
                                               tinfos,
                                               rkcg->rkcg_subscription,
                                               errored);


        /*
         * Propagate consumer errors for any non-existent or errored topics.
         * The function takes ownership of errored.
         */
        rd_kafka_propagate_consumer_topic_errors(
                rkcg, errored, "Subscribed topic not available");

        /*
         * Update (takes ownership of \c tinfos)
         */
        if (rd_kafka_cgrp_update_subscribed_topics(rkcg, tinfos) && do_join) {
                /* List of subscribed topics changed, trigger rejoin. */
                rd_kafka_dbg(rkcg->rkcg_rk,
                             CGRP|RD_KAFKA_DBG_METADATA|RD_KAFKA_DBG_CONSUMER,
                             "REJOIN",
                             "Group \"%.*s\": "
                             "subscription updated from metadata change: "
                             "rejoining group",
                             RD_KAFKAP_STR_PR(rkcg->rkcg_group_id));
                rd_kafka_cgrp_rejoin(rkcg);
        }
}
/**
 * @brief Check if the latest metadata affects the current subscription:
 * - matched topic added
 * - matched topic removed
 * - matched topic's partition count change
 *
 * @locks none
 * @locality rdkafka main thread
 */
@brief Check if the latest metadata affects the current subscription:
- matched topic added
- matched topic removed
- matched topic's partition count change

@locks none
@locality rdkafka main thread
@brief Check if the latest metadata affects the current subscription : - matched topic added - matched topic removed - matched topic ' s partition count change @locks none @locality rdkafka main thread
void rd_kafka_cgrp_metadata_update_check (rd_kafka_cgrp_t *rkcg, int do_join) {
        rd_list_t *tinfos;
        rd_kafka_topic_partition_list_t *errored;
        rd_kafka_assert(NULL, thrd_is_current(rkcg->rkcg_rk->rk_thread));
        if (!rkcg->rkcg_subscription || rkcg->rkcg_subscription->cnt == 0)
                return;
        errored = rd_kafka_topic_partition_list_new(0);
        tinfos = rd_list_new(rkcg->rkcg_subscription->cnt,
                             (void *)rd_kafka_topic_info_destroy);
        if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION)
                rd_kafka_metadata_topic_match(rkcg->rkcg_rk,
                                              tinfos, rkcg->rkcg_subscription,
                                              errored);
        else
                rd_kafka_metadata_topic_filter(rkcg->rkcg_rk,
                                               tinfos,
                                               rkcg->rkcg_subscription,
                                               errored);
        rd_kafka_propagate_consumer_topic_errors(
                rkcg, errored, "Subscribed topic not available");
        if (rd_kafka_cgrp_update_subscribed_topics(rkcg, tinfos) && do_join) {
                rd_kafka_dbg(rkcg->rkcg_rk,
                             CGRP|RD_KAFKA_DBG_METADATA|RD_KAFKA_DBG_CONSUMER,
                             "REJOIN",
                             "Group \"%.*s\": "
                             "subscription updated from metadata change: "
                             "rejoining group",
                             RD_KAFKAP_STR_PR(rkcg->rkcg_group_id));
                rd_kafka_cgrp_rejoin(rkcg);
        }
}
void rd_kafka_cgrp_metadata_update_check ( rd_kafka_cgrp_t * rkcg , int do_join ) { rd_list_t * tinfos ; rd_kafka_topic_partition_list_t * errored ; rd_kafka_assert ( NULL , thrd_is_current ( rkcg -> rkcg_rk -> rk_thread ) ) ; if ( ! rkcg -> rkcg_subscription || rkcg -> rkcg_subscription -> cnt == 0 ) return ; errored = rd_kafka_topic_partition_list_new ( 0 ) ; tinfos = rd_list_new ( rkcg -> rkcg_subscription -> cnt , ( void * ) rd_kafka_topic_info_destroy ) ; if ( rkcg -> rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION ) rd_kafka_metadata_topic_match ( rkcg -> rkcg_rk , tinfos , rkcg -> rkcg_subscription , errored ) ; else rd_kafka_metadata_topic_filter ( rkcg -> rkcg_rk , tinfos , rkcg -> rkcg_subscription , errored ) ; rd_kafka_propagate_consumer_topic_errors ( rkcg , errored , " " ) ; if ( rd_kafka_cgrp_update_subscribed_topics ( rkcg , tinfos ) && do_join ) { rd_kafka_dbg ( rkcg -> rkcg_rk , CGRP | RD_KAFKA_DBG_METADATA | RD_KAFKA_DBG_CONSUMER , " " , " \" \" " " " " " , RD_KAFKAP_STR_PR ( rkcg -> rkcg_group_id ) ) ; rd_kafka_cgrp_rejoin ( rkcg ) ; } }
@brief Check if the latest metadata affects the current subscription:
 - matched topic added
 - matched topic removed
 - matched topic's partition count change
@brief Check if the latest metadata affects the current subscription : - matched topic added - matched topic removed - matched topic ' s partition count change
/*
         * Unmatched topics will be added to the errored list.
         */ /*
         * Create a list of the topics in metadata that matches our subscription
         */ /*
         * Propagate consumer errors for any non-existent or errored topics.
         * The function takes ownership of errored.
         */ /*
         * Update (takes ownership of \c tinfos)
         */ /* List of subscribed topics changed, trigger rejoin. */