src/rdkafka_request.c
static void rd_kafka_handle_Metadata (rd_kafka_t *rk,
                                      rd_kafka_broker_t *rkb,
                                      rd_kafka_resp_err_t err,
                                      rd_kafka_buf_t *rkbuf,
                                      rd_kafka_buf_t *request,
                                      void *opaque) {
        rd_kafka_op_t *rko = opaque; /* Possibly NULL */
        struct rd_kafka_metadata *md = NULL;
        const rd_list_t *topics = request->rkbuf_u.Metadata.topics;
        int actions;

        rd_kafka_assert(NULL, err == RD_KAFKA_RESP_ERR__DESTROY ||
                        thrd_is_current(rk->rk_thread));

        /* Avoid metadata updates when we're terminating. */
        if (rd_kafka_terminating(rkb->rkb_rk) ||
            err == RD_KAFKA_RESP_ERR__DESTROY) {
                /* Terminating */
                goto done;
        }

        if (err)
                goto err;

        if (!topics)
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "===== Received metadata: %s =====",
                           request->rkbuf_u.Metadata.reason);
        else
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "===== Received metadata "
                           "(for %d requested topics): %s =====",
                           rd_list_cnt(topics),
                           request->rkbuf_u.Metadata.reason);

        err = rd_kafka_parse_Metadata(rkb, request, rkbuf, &md);
        if (err)
                goto err;

        if (rko && rko->rko_replyq.q) {
                /* Reply to metadata requester, passing on the metadata.
                 * Reuse requesting rko for the reply. */
                rko->rko_err = err;
                rko->rko_u.metadata.md = md;

                rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0);
                rko = NULL;
        } else {
                if (md)
                        rd_free(md);
        }

        goto done;

 err:
        actions = rd_kafka_err_action(
                rkb, err, request,

                RD_KAFKA_ERR_ACTION_RETRY,
                RD_KAFKA_RESP_ERR__PARTIAL,

                RD_KAFKA_ERR_ACTION_END);

        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return;
                /* FALLTHRU */
        } else {
                rd_rkb_log(rkb, LOG_WARNING, "METADATA",
                           "Metadata request failed: %s: %s (%dms): %s",
                           request->rkbuf_u.Metadata.reason,
                           rd_kafka_err2str(err),
                           (int)(request->rkbuf_ts_sent/1000),
                           rd_kafka_actions2str(actions));
        }



        /* FALLTHRU */

 done:
        if (rko)
                rd_kafka_op_destroy(rko);
}
/**
 * @brief Generic handler for Metadata responses
 *
 * @locality rdkafka main thread
 */
@brief Generic handler for Metadata responses
@locality rdkafka main thread
@brief Generic handler for Metadata responses @locality rdkafka main thread
static void rd_kafka_handle_Metadata (rd_kafka_t *rk,
                                      rd_kafka_broker_t *rkb,
                                      rd_kafka_resp_err_t err,
                                      rd_kafka_buf_t *rkbuf,
                                      rd_kafka_buf_t *request,
                                      void *opaque) {
        rd_kafka_op_t *rko = opaque; 
        struct rd_kafka_metadata *md = NULL;
        const rd_list_t *topics = request->rkbuf_u.Metadata.topics;
        int actions;
        rd_kafka_assert(NULL, err == RD_KAFKA_RESP_ERR__DESTROY ||
                        thrd_is_current(rk->rk_thread));
        if (rd_kafka_terminating(rkb->rkb_rk) ||
            err == RD_KAFKA_RESP_ERR__DESTROY) {
                goto done;
        }
        if (err)
                goto err;
        if (!topics)
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "===== Received metadata: %s =====",
                           request->rkbuf_u.Metadata.reason);
        else
                rd_rkb_dbg(rkb, METADATA, "METADATA",
                           "===== Received metadata "
                           "(for %d requested topics): %s =====",
                           rd_list_cnt(topics),
                           request->rkbuf_u.Metadata.reason);
        err = rd_kafka_parse_Metadata(rkb, request, rkbuf, &md);
        if (err)
                goto err;
        if (rko && rko->rko_replyq.q) {
                rko->rko_err = err;
                rko->rko_u.metadata.md = md;
                rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0);
                rko = NULL;
        } else {
                if (md)
                        rd_free(md);
        }
        goto done;
 err:
        actions = rd_kafka_err_action(
                rkb, err, request,
                RD_KAFKA_ERR_ACTION_RETRY,
                RD_KAFKA_RESP_ERR__PARTIAL,
                RD_KAFKA_ERR_ACTION_END);
        if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                if (rd_kafka_buf_retry(rkb, request))
                        return;
        } else {
                rd_rkb_log(rkb, LOG_WARNING, "METADATA",
                           "Metadata request failed: %s: %s (%dms): %s",
                           request->rkbuf_u.Metadata.reason,
                           rd_kafka_err2str(err),
                           (int)(request->rkbuf_ts_sent/1000),
                           rd_kafka_actions2str(actions));
        }
 done:
        if (rko)
                rd_kafka_op_destroy(rko);
}
static void rd_kafka_handle_Metadata ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * rkbuf , rd_kafka_buf_t * request , void * opaque ) { rd_kafka_op_t * rko = opaque ; struct rd_kafka_metadata * md = NULL ; const rd_list_t * topics = request -> rkbuf_u . Metadata . topics ; int actions ; rd_kafka_assert ( NULL , err == RD_KAFKA_RESP_ERR__DESTROY || thrd_is_current ( rk -> rk_thread ) ) ; if ( rd_kafka_terminating ( rkb -> rkb_rk ) || err == RD_KAFKA_RESP_ERR__DESTROY ) { goto done ; } if ( err ) goto err ; if ( ! topics ) rd_rkb_dbg ( rkb , METADATA , " " , " " , request -> rkbuf_u . Metadata . reason ) ; else rd_rkb_dbg ( rkb , METADATA , " " , " " " " , rd_list_cnt ( topics ) , request -> rkbuf_u . Metadata . reason ) ; err = rd_kafka_parse_Metadata ( rkb , request , rkbuf , & md ) ; if ( err ) goto err ; if ( rko && rko -> rko_replyq . q ) { rko -> rko_err = err ; rko -> rko_u . metadata . md = md ; rd_kafka_replyq_enq ( & rko -> rko_replyq , rko , 0 ) ; rko = NULL ; } else { if ( md ) rd_free ( md ) ; } goto done ; err : actions = rd_kafka_err_action ( rkb , err , request , RD_KAFKA_ERR_ACTION_RETRY , RD_KAFKA_RESP_ERR__PARTIAL , RD_KAFKA_ERR_ACTION_END ) ; if ( actions & RD_KAFKA_ERR_ACTION_RETRY ) { if ( rd_kafka_buf_retry ( rkb , request ) ) return ; } else { rd_rkb_log ( rkb , LOG_WARNING , " " , " " , request -> rkbuf_u . Metadata . reason , rd_kafka_err2str ( err ) , ( int ) ( request -> rkbuf_ts_sent / 1000 ) , rd_kafka_actions2str ( actions ) ) ; } done : if ( rko ) rd_kafka_op_destroy ( rko ) ; }
@brief Generic handler for Metadata responses
 @locality rdkafka main thread
@brief Generic handler for Metadata responses @locality rdkafka main thread
/* Possibly NULL */ /* Avoid metadata updates when we're terminating. */ /* Terminating */ /* Reply to metadata requester, passing on the metadata.
                 * Reuse requesting rko for the reply. */ /* FALLTHRU */ /* FALLTHRU */