plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_cgrp.c
static int rd_kafka_cgrp_metadata_refresh (rd_kafka_cgrp_t *rkcg,
                                            int *metadata_agep,
                                            const char *reason) {
        rd_kafka_t *rk = rkcg->rkcg_rk;
        rd_kafka_op_t *rko;
        rd_list_t topics;
        rd_kafka_resp_err_t err;

        rd_list_init(&topics, 8, rd_free);

        /* Insert all non-wildcard topics in cache. */
        rd_kafka_metadata_cache_hint_rktparlist(rkcg->rkcg_rk,
                                                rkcg->rkcg_subscription,
                                                NULL, 0/*dont replace*/);

        if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION) {
                /* For wildcard subscriptions make sure the
                 * cached full metadata isn't too old. */
                int metadata_age = -1;

                if (rk->rk_ts_full_metadata)
                        metadata_age = (int)(rd_clock() -
                                             rk->rk_ts_full_metadata)/1000;

                *metadata_agep = metadata_age;

                if (metadata_age != -1 &&
                    metadata_age <=
                    /* The +1000 is since metadata.refresh.interval.ms
                     * can be set to 0. */
                    rk->rk_conf.metadata_refresh_interval_ms + 1000) {
                        rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                                     "CGRPMETADATA",
                                     "%s: metadata for wildcard subscription "
                                     "is up to date (%dms old)",
                                     reason, *metadata_agep);
                        rd_list_destroy(&topics);
                        return 0; /* Up-to-date */
                }

        } else {
                /* Check that all subscribed topics are in the cache. */
                int r;

                rd_kafka_topic_partition_list_get_topic_names(
                        rkcg->rkcg_subscription, &topics, 0/*no regexps*/);

                rd_kafka_rdlock(rk);
                r = rd_kafka_metadata_cache_topics_count_exists(rk, &topics,
                                                                metadata_agep);
                rd_kafka_rdunlock(rk);

                if (r == rd_list_cnt(&topics)) {
                        rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                                     "CGRPMETADATA",
                                     "%s: metadata for subscription "
                                     "is up to date (%dms old)", reason,
                                     *metadata_agep);
                        rd_list_destroy(&topics);
                        return 0; /* Up-to-date and all topics exist. */
                }

                rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                             "CGRPMETADATA",
                             "%s: metadata for subscription "
                             "only available for %d/%d topics (%dms old)",
                             reason, r, rd_list_cnt(&topics), *metadata_agep);

        }

        /* Async request, result will be triggered from
         * rd_kafka_parse_metadata(). */
        rko = rd_kafka_op_new_cb(rkcg->rkcg_rk, RD_KAFKA_OP_METADATA,
                                 rd_kafka_cgrp_handle_Metadata_op);
        rd_kafka_op_set_replyq(rko, rkcg->rkcg_ops, 0);

        err = rd_kafka_metadata_request(rkcg->rkcg_rk, NULL, &topics,
                                        rd_true/*cgrp_update*/,
                                        reason, rko);
        if (err) {
                rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                             "CGRPMETADATA",
                             "%s: need to refresh metadata (%dms old) "
                             "but no usable brokers available: %s",
                             reason, *metadata_agep, rd_kafka_err2str(err));
                rd_kafka_op_destroy(rko);
        }

        rd_list_destroy(&topics);

        return err ? -1 : 1;
}
/**
 * @brief (Async) Refresh metadata (for cgrp's needs)
 *
 * @returns 1 if metadata refresh was requested, or 0 if metadata is
 *          up to date, or -1 if no broker is available for metadata requests.
 *
 * @locks none
 * @locality rdkafka main thread
 */
@brief (Async) Refresh metadata (for cgrp's needs)
@returns 1 if metadata refresh was requested, or 0 if metadata is
up to date, or -1 if no broker is available for metadata requests.

@locks none
@locality rdkafka main thread
@brief ( Async ) Refresh metadata ( for cgrp ' s needs ) @returns 1 if metadata refresh was requested or 0 if metadata is up to date or - 1 if no broker is available for metadata requests . @locks none @locality rdkafka main thread
static int rd_kafka_cgrp_metadata_refresh (rd_kafka_cgrp_t *rkcg,
                                            int *metadata_agep,
                                            const char *reason) {
        rd_kafka_t *rk = rkcg->rkcg_rk;
        rd_kafka_op_t *rko;
        rd_list_t topics;
        rd_kafka_resp_err_t err;
        rd_list_init(&topics, 8, rd_free);
        rd_kafka_metadata_cache_hint_rktparlist(rkcg->rkcg_rk,
                                                rkcg->rkcg_subscription,
                                                NULL, 0);
        if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION) {
                int metadata_age = -1;
                if (rk->rk_ts_full_metadata)
                        metadata_age = (int)(rd_clock() -
                                             rk->rk_ts_full_metadata)/1000;
                *metadata_agep = metadata_age;
                if (metadata_age != -1 &&
                    metadata_age <=
                    rk->rk_conf.metadata_refresh_interval_ms + 1000) {
                        rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                                     "CGRPMETADATA",
                                     "%s: metadata for wildcard subscription "
                                     "is up to date (%dms old)",
                                     reason, *metadata_agep);
                        rd_list_destroy(&topics);
                        return 0; 
                }
        } else {
                int r;
                rd_kafka_topic_partition_list_get_topic_names(
                        rkcg->rkcg_subscription, &topics, 0);
                rd_kafka_rdlock(rk);
                r = rd_kafka_metadata_cache_topics_count_exists(rk, &topics,
                                                                metadata_agep);
                rd_kafka_rdunlock(rk);
                if (r == rd_list_cnt(&topics)) {
                        rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                                     "CGRPMETADATA",
                                     "%s: metadata for subscription "
                                     "is up to date (%dms old)", reason,
                                     *metadata_agep);
                        rd_list_destroy(&topics);
                        return 0; 
                }
                rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                             "CGRPMETADATA",
                             "%s: metadata for subscription "
                             "only available for %d/%d topics (%dms old)",
                             reason, r, rd_list_cnt(&topics), *metadata_agep);
        }
        rko = rd_kafka_op_new_cb(rkcg->rkcg_rk, RD_KAFKA_OP_METADATA,
                                 rd_kafka_cgrp_handle_Metadata_op);
        rd_kafka_op_set_replyq(rko, rkcg->rkcg_ops, 0);
        err = rd_kafka_metadata_request(rkcg->rkcg_rk, NULL, &topics,
                                        rd_true,
                                        reason, rko);
        if (err) {
                rd_kafka_dbg(rk, CGRP|RD_KAFKA_DBG_METADATA,
                             "CGRPMETADATA",
                             "%s: need to refresh metadata (%dms old) "
                             "but no usable brokers available: %s",
                             reason, *metadata_agep, rd_kafka_err2str(err));
                rd_kafka_op_destroy(rko);
        }
        rd_list_destroy(&topics);
        return err ? -1 : 1;
}
static int rd_kafka_cgrp_metadata_refresh ( rd_kafka_cgrp_t * rkcg , int * metadata_agep , const char * reason ) { rd_kafka_t * rk = rkcg -> rkcg_rk ; rd_kafka_op_t * rko ; rd_list_t topics ; rd_kafka_resp_err_t err ; rd_list_init ( & topics , 8 , rd_free ) ; rd_kafka_metadata_cache_hint_rktparlist ( rkcg -> rkcg_rk , rkcg -> rkcg_subscription , NULL , 0 ) ; if ( rkcg -> rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION ) { int metadata_age = -1 ; if ( rk -> rk_ts_full_metadata ) metadata_age = ( int ) ( rd_clock ( ) - rk -> rk_ts_full_metadata ) / 1000 ; * metadata_agep = metadata_age ; if ( metadata_age != -1 && metadata_age <= rk -> rk_conf . metadata_refresh_interval_ms + 1000 ) { rd_kafka_dbg ( rk , CGRP | RD_KAFKA_DBG_METADATA , " " , " " " " , reason , * metadata_agep ) ; rd_list_destroy ( & topics ) ; return 0 ; } } else { int r ; rd_kafka_topic_partition_list_get_topic_names ( rkcg -> rkcg_subscription , & topics , 0 ) ; rd_kafka_rdlock ( rk ) ; r = rd_kafka_metadata_cache_topics_count_exists ( rk , & topics , metadata_agep ) ; rd_kafka_rdunlock ( rk ) ; if ( r == rd_list_cnt ( & topics ) ) { rd_kafka_dbg ( rk , CGRP | RD_KAFKA_DBG_METADATA , " " , " " " " , reason , * metadata_agep ) ; rd_list_destroy ( & topics ) ; return 0 ; } rd_kafka_dbg ( rk , CGRP | RD_KAFKA_DBG_METADATA , " " , " " " " , reason , r , rd_list_cnt ( & topics ) , * metadata_agep ) ; } rko = rd_kafka_op_new_cb ( rkcg -> rkcg_rk , RD_KAFKA_OP_METADATA , rd_kafka_cgrp_handle_Metadata_op ) ; rd_kafka_op_set_replyq ( rko , rkcg -> rkcg_ops , 0 ) ; err = rd_kafka_metadata_request ( rkcg -> rkcg_rk , NULL , & topics , rd_true , reason , rko ) ; if ( err ) { rd_kafka_dbg ( rk , CGRP | RD_KAFKA_DBG_METADATA , " " , " " " " , reason , * metadata_agep , rd_kafka_err2str ( err ) ) ; rd_kafka_op_destroy ( rko ) ; } rd_list_destroy ( & topics ) ; return err ? -1 : 1 ; }
@brief (Async) Refresh metadata (for cgrp's needs)
 @returns 1 if metadata refresh was requested, or 0 if metadata is
          up to date, or -1 if no broker is available for metadata requests.
@brief ( Async ) Refresh metadata ( for cgrp ' s needs ) @returns 1 if metadata refresh was requested or 0 if metadata is up to date or - 1 if no broker is available for metadata requests .
/* Insert all non-wildcard topics in cache. */ /*dont replace*/ /* For wildcard subscriptions make sure the
                 * cached full metadata isn't too old. */ /* The +1000 is since metadata.refresh.interval.ms
                     * can be set to 0. */ /* Up-to-date */ /* Check that all subscribed topics are in the cache. */ /*no regexps*/ /* Up-to-date and all topics exist. */ /* Async request, result will be triggered from
         * rd_kafka_parse_metadata(). */ /*cgrp_update*/