plugins/out_kafka/librdkafka-1.5.0/src/rdkafka_admin.c
static void rd_kafka_admin_handle_response (rd_kafka_t *rk,
                                            rd_kafka_broker_t *rkb,
                                            rd_kafka_resp_err_t err,
                                            rd_kafka_buf_t *reply,
                                            rd_kafka_buf_t *request,
                                            void *opaque) {
        rd_kafka_enq_once_t *eonce = opaque;
        rd_kafka_op_t *rko;

        /* From ...add_source("send") */
        rko = rd_kafka_enq_once_disable(eonce);

        if (!rko) {
                /* The operation timed out and the worker was
                 * dismantled while we were waiting for broker response,
                 * do nothing - everything has been cleaned up. */
                rd_kafka_dbg(rk, ADMIN, "ADMIN",
                             "Dropping outdated %sResponse with return code %s",
                             request ?
                             rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey):
                             "???",
                             rd_kafka_err2str(err));
                return;
        }

        /* Attach reply buffer to rko for parsing in the worker. */
        rd_assert(!rko->rko_u.admin_request.reply_buf);
        rko->rko_u.admin_request.reply_buf = reply;
        rko->rko_err = err;

        if (rko->rko_op_cb(rk, NULL, rko) == RD_KAFKA_OP_RES_HANDLED)
                rd_kafka_op_destroy(rko);

}
/**
 * @brief Handle response from broker by triggering worker callback.
 *
 * @param opaque is the eonce from the worker protocol request call.
 */
@brief Handle response from broker by triggering worker callback.
@param opaque is the eonce from the worker protocol request call.
@brief Handle response from broker by triggering worker callback . @param opaque is the eonce from the worker protocol request call .
static void rd_kafka_admin_handle_response (rd_kafka_t *rk,
                                            rd_kafka_broker_t *rkb,
                                            rd_kafka_resp_err_t err,
                                            rd_kafka_buf_t *reply,
                                            rd_kafka_buf_t *request,
                                            void *opaque) {
        rd_kafka_enq_once_t *eonce = opaque;
        rd_kafka_op_t *rko;
        rko = rd_kafka_enq_once_disable(eonce);
        if (!rko) {
                rd_kafka_dbg(rk, ADMIN, "ADMIN",
                             "Dropping outdated %sResponse with return code %s",
                             request ?
                             rd_kafka_ApiKey2str(request->rkbuf_reqhdr.ApiKey):
                             "???",
                             rd_kafka_err2str(err));
                return;
        }
        rd_assert(!rko->rko_u.admin_request.reply_buf);
        rko->rko_u.admin_request.reply_buf = reply;
        rko->rko_err = err;
        if (rko->rko_op_cb(rk, NULL, rko) == RD_KAFKA_OP_RES_HANDLED)
                rd_kafka_op_destroy(rko);
}
static void rd_kafka_admin_handle_response ( rd_kafka_t * rk , rd_kafka_broker_t * rkb , rd_kafka_resp_err_t err , rd_kafka_buf_t * reply , rd_kafka_buf_t * request , void * opaque ) { rd_kafka_enq_once_t * eonce = opaque ; rd_kafka_op_t * rko ; rko = rd_kafka_enq_once_disable ( eonce ) ; if ( ! rko ) { rd_kafka_dbg ( rk , ADMIN , " " , " " , request ? rd_kafka_ApiKey2str ( request -> rkbuf_reqhdr . ApiKey ) : " " , rd_kafka_err2str ( err ) ) ; return ; } rd_assert ( ! rko -> rko_u . admin_request . reply_buf ) ; rko -> rko_u . admin_request . reply_buf = reply ; rko -> rko_err = err ; if ( rko -> rko_op_cb ( rk , NULL , rko ) == RD_KAFKA_OP_RES_HANDLED ) rd_kafka_op_destroy ( rko ) ; }
@brief Handle response from broker by triggering worker callback.
@brief Handle response from broker by triggering worker callback .
/* From ...add_source("send") */ /* The operation timed out and the worker was
                 * dismantled while we were waiting for broker response,
                 * do nothing - everything has been cleaned up. */ /* Attach reply buffer to rko for parsing in the worker. */