src/rdkafka_request.c
rd_kafka_resp_err_t
rd_kafka_MetadataRequest (rd_kafka_broker_t *rkb,
                          const rd_list_t *topics, const char *reason,
                          rd_kafka_op_t *rko) {
        rd_kafka_buf_t *rkbuf;
        int16_t ApiVersion = 0;
        int features;
        int topic_cnt = topics ? rd_list_cnt(topics) : 0;
        int *full_incr = NULL;

        ApiVersion = rd_kafka_broker_ApiVersion_supported(rkb,
                                                          RD_KAFKAP_Metadata,
                                                          0, 2,
                                                          &features);

        rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_Metadata, 1,
                                         4 + (50 * topic_cnt));

        if (!reason)
                reason = "";

        rkbuf->rkbuf_u.Metadata.reason = rd_strdup(reason);

        if (!topics && ApiVersion >= 1) {
                /* a null(0) array (in the protocol) represents no topics */
                rd_kafka_buf_write_i32(rkbuf, 0);
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "Request metadata for brokers only: %s", reason);
                full_incr = &rkb->rkb_rk->rk_metadata_cache.
                        rkmc_full_brokers_sent;

        } else {
                if (topic_cnt == 0 && !rko)
                        full_incr = &rkb->rkb_rk->rk_metadata_cache.
                                rkmc_full_topics_sent;

                if (topic_cnt == 0 && ApiVersion >= 1)
                        rd_kafka_buf_write_i32(rkbuf, -1); /* Null: all topics*/
                else
                        rd_kafka_buf_write_i32(rkbuf, topic_cnt);

                if (topic_cnt == 0) {
                        rkbuf->rkbuf_u.Metadata.all_topics = 1;
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Request metadata for all topics: "
                                   "%s", reason);
                } else
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Request metadata for %d topic(s): "
                                   "%s", topic_cnt, reason);
        }

        if (full_incr) {
                /* Avoid multiple outstanding full requests
                 * (since they are redundant and side-effect-less).
                 * Forced requests (app using metadata() API) are passed
                 * through regardless. */

                mtx_lock(&rkb->rkb_rk->rk_metadata_cache.
                         rkmc_full_lock);
                if (*full_incr > 0 && (!rko || !rko->rko_u.metadata.force)) {
                        mtx_unlock(&rkb->rkb_rk->rk_metadata_cache.
                                   rkmc_full_lock);
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Skipping metadata request: %s: "
                                   "full request already in-transit",
                                   reason);
                        rd_kafka_buf_destroy(rkbuf);
                        return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS;
                }

                (*full_incr)++;
                mtx_unlock(&rkb->rkb_rk->rk_metadata_cache.
                           rkmc_full_lock);
                rkbuf->rkbuf_u.Metadata.decr = full_incr;
                rkbuf->rkbuf_u.Metadata.decr_lock = &rkb->rkb_rk->
                        rk_metadata_cache.rkmc_full_lock;
        }


        if (topic_cnt > 0) {
                char *topic;
                int i;

                /* Maintain a copy of the topics list so we can purge
                 * hints from the metadata cache on error. */
                rkbuf->rkbuf_u.Metadata.topics =
                        rd_list_copy(topics, rd_list_string_copy, NULL);

                RD_LIST_FOREACH(topic, topics, i)
                        rd_kafka_buf_write_str(rkbuf, topic, -1);

        }

        rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0);

        /* Metadata requests are part of the important control plane
         * and should go before most other requests (Produce, Fetch, etc). */
        rkbuf->rkbuf_prio = RD_KAFKA_PRIO_HIGH;

        rd_kafka_broker_buf_enq_replyq(rkb, rkbuf,
                                       /* Handle response thru rk_ops,
                                        * but forward parsed result to
                                        * rko's replyq when done. */
                                       RD_KAFKA_REPLYQ(rkb->rkb_rk->
                                                       rk_ops, 0),
                                       rd_kafka_handle_Metadata, rko);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
/**
 * @brief Construct MetadataRequest (does not send)
 *
 * \p topics is a list of topic names (char *) to request.
 *
 * !topics          - only request brokers (if supported by broker, else
 *                    all topics)
 *  topics.cnt==0   - all topics in cluster are requested
 *  topics.cnt >0   - only specified topics are requested
 *
 * @param reason    - metadata request reason
 * @param rko       - (optional) rko with replyq for handling response.
 *                    Specifying an rko forces a metadata request even if
 *                    there is already a matching one in-transit.
 *
 * If full metadata for all topics is requested (or all brokers, which
 * results in all-topics on older brokers) and there is already a full request
 * in transit then this function will return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS
 * otherwise RD_KAFKA_RESP_ERR_NO_ERROR. If \p rko is non-NULL the request
 * is sent regardless.
 */
@brief Construct MetadataRequest (does not send)
\p topics is a list of topic names (char *) to request.

!topics          - only request brokers (if supported by broker, else
all topics)
topics.cnt==0   - all topics in cluster are requested
topics.cnt >0   - only specified topics are requested

@param reason    - metadata request reason
@param rko       - (optional) rko with replyq for handling response.
Specifying an rko forces a metadata request even if
there is already a matching one in-transit.

If full metadata for all topics is requested (or all brokers, which
results in all-topics on older brokers) and there is already a full request
in transit then this function will return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS
otherwise RD_KAFKA_RESP_ERR_NO_ERROR. If \p rko is non-NULL the request
is sent regardless.
@brief Construct MetadataRequest ( does not send ) \ p topics is a list of topic names ( char * ) to request . !topics - only request brokers ( if supported by broker else all topics ) topics . cnt == 0 - all topics in cluster are requested topics . cnt > 0 - only specified topics are requested @param reason - metadata request reason @param rko - ( optional ) rko with replyq for handling response . Specifying an rko forces a metadata request even if there is already a matching one in - transit . If full metadata for all topics is requested ( or all brokers which results in all - topics on older brokers ) and there is already a full request in transit then this function will return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS otherwise RD_KAFKA_RESP_ERR_NO_ERROR . If \ p rko is non - NULL the request is sent regardless .
rd_kafka_resp_err_t
rd_kafka_MetadataRequest (rd_kafka_broker_t *rkb,
                          const rd_list_t *topics, const char *reason,
                          rd_kafka_op_t *rko) {
        rd_kafka_buf_t *rkbuf;
        int16_t ApiVersion = 0;
        int features;
        int topic_cnt = topics ? rd_list_cnt(topics) : 0;
        int *full_incr = NULL;
        ApiVersion = rd_kafka_broker_ApiVersion_supported(rkb,
                                                          RD_KAFKAP_Metadata,
                                                          0, 2,
                                                          &features);
        rkbuf = rd_kafka_buf_new_request(rkb, RD_KAFKAP_Metadata, 1,
                                         4 + (50 * topic_cnt));
        if (!reason)
                reason = "";
        rkbuf->rkbuf_u.Metadata.reason = rd_strdup(reason);
        if (!topics && ApiVersion >= 1) {
                rd_kafka_buf_write_i32(rkbuf, 0);
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "Request metadata for brokers only: %s", reason);
                full_incr = &rkb->rkb_rk->rk_metadata_cache.
                        rkmc_full_brokers_sent;
        } else {
                if (topic_cnt == 0 && !rko)
                        full_incr = &rkb->rkb_rk->rk_metadata_cache.
                                rkmc_full_topics_sent;
                if (topic_cnt == 0 && ApiVersion >= 1)
                        rd_kafka_buf_write_i32(rkbuf, -1); 
                else
                        rd_kafka_buf_write_i32(rkbuf, topic_cnt);
                if (topic_cnt == 0) {
                        rkbuf->rkbuf_u.Metadata.all_topics = 1;
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Request metadata for all topics: "
                                   "%s", reason);
                } else
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Request metadata for %d topic(s): "
                                   "%s", topic_cnt, reason);
        }
        if (full_incr) {
                mtx_lock(&rkb->rkb_rk->rk_metadata_cache.
                         rkmc_full_lock);
                if (*full_incr > 0 && (!rko || !rko->rko_u.metadata.force)) {
                        mtx_unlock(&rkb->rkb_rk->rk_metadata_cache.
                                   rkmc_full_lock);
                        rd_rkb_dbg(rkb, METADATA, "METADATA",
                                   "Skipping metadata request: %s: "
                                   "full request already in-transit",
                                   reason);
                        rd_kafka_buf_destroy(rkbuf);
                        return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS;
                }
                (*full_incr)++;
                mtx_unlock(&rkb->rkb_rk->rk_metadata_cache.
                           rkmc_full_lock);
                rkbuf->rkbuf_u.Metadata.decr = full_incr;
                rkbuf->rkbuf_u.Metadata.decr_lock = &rkb->rkb_rk->
                        rk_metadata_cache.rkmc_full_lock;
        }
        if (topic_cnt > 0) {
                char *topic;
                int i;
                rkbuf->rkbuf_u.Metadata.topics =
                        rd_list_copy(topics, rd_list_string_copy, NULL);
                RD_LIST_FOREACH(topic, topics, i)
                        rd_kafka_buf_write_str(rkbuf, topic, -1);
        }
        rd_kafka_buf_ApiVersion_set(rkbuf, ApiVersion, 0);
        rkbuf->rkbuf_prio = RD_KAFKA_PRIO_HIGH;
        rd_kafka_broker_buf_enq_replyq(rkb, rkbuf,
                                       RD_KAFKA_REPLYQ(rkb->rkb_rk->
                                                       rk_ops, 0),
                                       rd_kafka_handle_Metadata, rko);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}
rd_kafka_resp_err_t rd_kafka_MetadataRequest ( rd_kafka_broker_t * rkb , const rd_list_t * topics , const char * reason , rd_kafka_op_t * rko ) { rd_kafka_buf_t * rkbuf ; int16_t ApiVersion = 0 ; int features ; int topic_cnt = topics ? rd_list_cnt ( topics ) : 0 ; int * full_incr = NULL ; ApiVersion = rd_kafka_broker_ApiVersion_supported ( rkb , RD_KAFKAP_Metadata , 0 , 2 , & features ) ; rkbuf = rd_kafka_buf_new_request ( rkb , RD_KAFKAP_Metadata , 1 , 4 + ( 50 * topic_cnt ) ) ; if ( ! reason ) reason = " " ; rkbuf -> rkbuf_u . Metadata . reason = rd_strdup ( reason ) ; if ( ! topics && ApiVersion >= 1 ) { rd_kafka_buf_write_i32 ( rkbuf , 0 ) ; rd_rkb_dbg ( rkb , METADATA , " " , " " , reason ) ; full_incr = & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_brokers_sent ; } else { if ( topic_cnt == 0 && ! rko ) full_incr = & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_topics_sent ; if ( topic_cnt == 0 && ApiVersion >= 1 ) rd_kafka_buf_write_i32 ( rkbuf , -1 ) ; else rd_kafka_buf_write_i32 ( rkbuf , topic_cnt ) ; if ( topic_cnt == 0 ) { rkbuf -> rkbuf_u . Metadata . all_topics = 1 ; rd_rkb_dbg ( rkb , METADATA , " " , " " " " , reason ) ; } else rd_rkb_dbg ( rkb , METADATA , " " , " " " " , topic_cnt , reason ) ; } if ( full_incr ) { mtx_lock ( & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_lock ) ; if ( * full_incr > 0 && ( ! rko || ! rko -> rko_u . metadata . force ) ) { mtx_unlock ( & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_lock ) ; rd_rkb_dbg ( rkb , METADATA , " " , " " " " , reason ) ; rd_kafka_buf_destroy ( rkbuf ) ; return RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS ; } ( * full_incr ) ++ ; mtx_unlock ( & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_lock ) ; rkbuf -> rkbuf_u . Metadata . decr = full_incr ; rkbuf -> rkbuf_u . Metadata . decr_lock = & rkb -> rkb_rk -> rk_metadata_cache . rkmc_full_lock ; } if ( topic_cnt > 0 ) { char * topic ; int i ; rkbuf -> rkbuf_u . Metadata . topics = rd_list_copy ( topics , rd_list_string_copy , NULL ) ; RD_LIST_FOREACH ( topic , topics , i ) rd_kafka_buf_write_str ( rkbuf , topic , -1 ) ; } rd_kafka_buf_ApiVersion_set ( rkbuf , ApiVersion , 0 ) ; rkbuf -> rkbuf_prio = RD_KAFKA_PRIO_HIGH ; rd_kafka_broker_buf_enq_replyq ( rkb , rkbuf , RD_KAFKA_REPLYQ ( rkb -> rkb_rk -> rk_ops , 0 ) , rd_kafka_handle_Metadata , rko ) ; return RD_KAFKA_RESP_ERR_NO_ERROR ; }
@brief Construct MetadataRequest (does not send)
 \p topics is a list of topic names (char *) to request.
@brief Construct MetadataRequest ( does not send ) \ p topics is a list of topic names ( char * ) to request .
/* a null(0) array (in the protocol) represents no topics */ /* Null: all topics*/ /* Avoid multiple outstanding full requests
                 * (since they are redundant and side-effect-less).
                 * Forced requests (app using metadata() API) are passed
                 * through regardless. */ /* Maintain a copy of the topics list so we can purge
                 * hints from the metadata cache on error. */ /* Metadata requests are part of the important control plane
         * and should go before most other requests (Produce, Fetch, etc). */ /* Handle response thru rk_ops,
                                        * but forward parsed result to
                                        * rko's replyq when done. */