/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2012-2022, Magnus Edenhill
 *               2023, Confluent Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <stdarg.h>

#include "rdkafka_int.h"
#include "rdkafka_op.h"
#include "rdkafka_topic.h"
#include "rdkafka_partition.h"
#include "rdkafka_proto.h"
#include "rdkafka_offset.h"
#include "rdkafka_error.h"

/* Current number of rd_kafka_op_t */
rd_atomic32_t rd_kafka_op_cnt;


const char *rd_kafka_op2str(rd_kafka_op_type_t type) {
        int skiplen                                = 6;
        static const char *names[RD_KAFKA_OP__END] = {
            [RD_KAFKA_OP_NONE]             = "REPLY:NONE",
            [RD_KAFKA_OP_FETCH]            = "REPLY:FETCH",
            [RD_KAFKA_OP_ERR]              = "REPLY:ERR",
            [RD_KAFKA_OP_CONSUMER_ERR]     = "REPLY:CONSUMER_ERR",
            [RD_KAFKA_OP_DR]               = "REPLY:DR",
            [RD_KAFKA_OP_STATS]            = "REPLY:STATS",
            [RD_KAFKA_OP_OFFSET_COMMIT]    = "REPLY:OFFSET_COMMIT",
            [RD_KAFKA_OP_NODE_UPDATE]      = "REPLY:NODE_UPDATE",
            [RD_KAFKA_OP_XMIT_BUF]         = "REPLY:XMIT_BUF",
            [RD_KAFKA_OP_RECV_BUF]         = "REPLY:RECV_BUF",
            [RD_KAFKA_OP_XMIT_RETRY]       = "REPLY:XMIT_RETRY",
            [RD_KAFKA_OP_FETCH_START]      = "REPLY:FETCH_START",
            [RD_KAFKA_OP_FETCH_STOP]       = "REPLY:FETCH_STOP",
            [RD_KAFKA_OP_SEEK]             = "REPLY:SEEK",
            [RD_KAFKA_OP_PAUSE]            = "REPLY:PAUSE",
            [RD_KAFKA_OP_OFFSET_FETCH]     = "REPLY:OFFSET_FETCH",
            [RD_KAFKA_OP_PARTITION_JOIN]   = "REPLY:PARTITION_JOIN",
            [RD_KAFKA_OP_PARTITION_LEAVE]  = "REPLY:PARTITION_LEAVE",
            [RD_KAFKA_OP_REBALANCE]        = "REPLY:REBALANCE",
            [RD_KAFKA_OP_TERMINATE]        = "REPLY:TERMINATE",
            [RD_KAFKA_OP_COORD_QUERY]      = "REPLY:COORD_QUERY",
            [RD_KAFKA_OP_SUBSCRIBE]        = "REPLY:SUBSCRIBE",
            [RD_KAFKA_OP_ASSIGN]           = "REPLY:ASSIGN",
            [RD_KAFKA_OP_GET_SUBSCRIPTION] = "REPLY:GET_SUBSCRIPTION",
            [RD_KAFKA_OP_GET_ASSIGNMENT]   = "REPLY:GET_ASSIGNMENT",
            [RD_KAFKA_OP_THROTTLE]         = "REPLY:THROTTLE",
            [RD_KAFKA_OP_NAME]             = "REPLY:NAME",
            [RD_KAFKA_OP_CG_METADATA]      = "REPLY:CG_METADATA",
            [RD_KAFKA_OP_OFFSET_RESET]     = "REPLY:OFFSET_RESET",
            [RD_KAFKA_OP_METADATA]         = "REPLY:METADATA",
            [RD_KAFKA_OP_LOG]              = "REPLY:LOG",
            [RD_KAFKA_OP_WAKEUP]           = "REPLY:WAKEUP",
            [RD_KAFKA_OP_CREATETOPICS]     = "REPLY:CREATETOPICS",
            [RD_KAFKA_OP_DELETETOPICS]     = "REPLY:DELETETOPICS",
            [RD_KAFKA_OP_CREATEPARTITIONS] = "REPLY:CREATEPARTITIONS",
            [RD_KAFKA_OP_ALTERCONFIGS]     = "REPLY:ALTERCONFIGS",
            [RD_KAFKA_OP_INCREMENTALALTERCONFIGS] =
                "REPLY:INCREMENTALALTERCONFIGS",
            [RD_KAFKA_OP_DESCRIBECONFIGS]    = "REPLY:DESCRIBECONFIGS",
            [RD_KAFKA_OP_DELETERECORDS]      = "REPLY:DELETERECORDS",
            [RD_KAFKA_OP_LISTCONSUMERGROUPS] = "REPLY:LISTCONSUMERGROUPS",
            [RD_KAFKA_OP_DESCRIBECONSUMERGROUPS] =
                "REPLY:DESCRIBECONSUMERGROUPS",
            [RD_KAFKA_OP_DESCRIBETOPICS]  = "REPLY:DESCRIBETOPICS",
            [RD_KAFKA_OP_DESCRIBECLUSTER] = "REPLY:DESCRIBECLUSTER",
            [RD_KAFKA_OP_DELETEGROUPS]    = "REPLY:DELETEGROUPS",
            [RD_KAFKA_OP_DELETECONSUMERGROUPOFFSETS] =
                "REPLY:DELETECONSUMERGROUPOFFSETS",
            [RD_KAFKA_OP_CREATEACLS]   = "REPLY:CREATEACLS",
            [RD_KAFKA_OP_DESCRIBEACLS] = "REPLY:DESCRIBEACLS",
            [RD_KAFKA_OP_DELETEACLS]   = "REPLY:DELETEACLS",
            [RD_KAFKA_OP_ALTERCONSUMERGROUPOFFSETS] =
                "REPLY:ALTERCONSUMERGROUPOFFSETS",
            [RD_KAFKA_OP_LISTCONSUMERGROUPOFFSETS] =
                "REPLY:LISTCONSUMERGROUPOFFSETS",
            [RD_KAFKA_OP_ADMIN_FANOUT]        = "REPLY:ADMIN_FANOUT",
            [RD_KAFKA_OP_ADMIN_RESULT]        = "REPLY:ADMIN_RESULT",
            [RD_KAFKA_OP_PURGE]               = "REPLY:PURGE",
            [RD_KAFKA_OP_CONNECT]             = "REPLY:CONNECT",
            [RD_KAFKA_OP_OAUTHBEARER_REFRESH] = "REPLY:OAUTHBEARER_REFRESH",
            [RD_KAFKA_OP_MOCK]                = "REPLY:MOCK",
            [RD_KAFKA_OP_BROKER_MONITOR]      = "REPLY:BROKER_MONITOR",
            [RD_KAFKA_OP_TXN]                 = "REPLY:TXN",
            [RD_KAFKA_OP_GET_REBALANCE_PROTOCOL] =
                "REPLY:GET_REBALANCE_PROTOCOL",
            [RD_KAFKA_OP_LEADERS]     = "REPLY:LEADERS",
            [RD_KAFKA_OP_BARRIER]     = "REPLY:BARRIER",
            [RD_KAFKA_OP_SASL_REAUTH] = "REPLY:SASL_REAUTH",
            [RD_KAFKA_OP_ALTERUSERSCRAMCREDENTIALS] =
                "REPLY:ALTERUSERSCRAMCREDENTIALS",
            [RD_KAFKA_OP_DESCRIBEUSERSCRAMCREDENTIALS] =
                "REPLY:DESCRIBEUSERSCRAMCREDENTIALS",
            [RD_KAFKA_OP_LISTOFFSETS] = "REPLY:LISTOFFSETS",
        };

        if (type & RD_KAFKA_OP_REPLY)
                skiplen = 0;

        rd_assert((names[type & ~RD_KAFKA_OP_FLAGMASK] != NULL) ||
                  !*"add OP type to rd_kafka_op2str()");
        return names[type & ~RD_KAFKA_OP_FLAGMASK] + skiplen;
}


void rd_kafka_op_print(FILE *fp, const char *prefix, rd_kafka_op_t *rko) {
        fprintf(fp,
                "%s((rd_kafka_op_t*)%p)\n"
                "%s Type: %s (0x%x), Version: %" PRId32 "\n",
                prefix, rko, prefix, rd_kafka_op2str(rko->rko_type),
                rko->rko_type, rko->rko_version);
        if (rko->rko_err)
                fprintf(fp, "%s Error: %s\n", prefix,
                        rd_kafka_err2str(rko->rko_err));
        if (rko->rko_replyq.q)
                fprintf(fp, "%s Replyq %p v%d (%s)\n", prefix,
                        rko->rko_replyq.q, rko->rko_replyq.version,
#if ENABLE_DEVEL
                        rko->rko_replyq._id
#else
                        ""
#endif
                );
        if (rko->rko_rktp) {
                fprintf(fp,
                        "%s ((rd_kafka_toppar_t*)%p) "
                        "%s [%" PRId32 "] v%d\n",
                        prefix, rko->rko_rktp,
                        rko->rko_rktp->rktp_rkt->rkt_topic->str,
                        rko->rko_rktp->rktp_partition,
                        rd_atomic32_get(&rko->rko_rktp->rktp_version));
        }

        switch (rko->rko_type & ~RD_KAFKA_OP_FLAGMASK) {
        case RD_KAFKA_OP_FETCH:
                fprintf(fp, "%s Offset: %" PRId64 "\n", prefix,
                        rko->rko_u.fetch.rkm.rkm_offset);
                break;
        case RD_KAFKA_OP_CONSUMER_ERR:
                fprintf(fp, "%s Offset: %" PRId64 "\n", prefix,
                        rko->rko_u.err.offset);
                /* FALLTHRU */
        case RD_KAFKA_OP_ERR:
                fprintf(fp, "%s Reason: %s\n", prefix, rko->rko_u.err.errstr);
                break;
        case RD_KAFKA_OP_DR:
                fprintf(fp, "%s %" PRId32 " messages on %s\n", prefix,
                        rko->rko_u.dr.msgq.rkmq_msg_cnt,
                        rko->rko_u.dr.rkt ? rko->rko_u.dr.rkt->rkt_topic->str
                                          : "(n/a)");
                break;
        case RD_KAFKA_OP_OFFSET_COMMIT:
                fprintf(fp, "%s Callback: %p (opaque %p)\n", prefix,
                        rko->rko_u.offset_commit.cb,
                        rko->rko_u.offset_commit.opaque);
                fprintf(fp, "%s %d partitions\n", prefix,
                        rko->rko_u.offset_commit.partitions
                            ? rko->rko_u.offset_commit.partitions->cnt
                            : 0);
                break;

        case RD_KAFKA_OP_LOG:
                fprintf(fp, "%s Log: %%%d %s: %s\n", prefix,
                        rko->rko_u.log.level, rko->rko_u.log.fac,
                        rko->rko_u.log.str);
                break;

        default:
                break;
        }
}


rd_kafka_op_t *rd_kafka_op_new0(const char *source, rd_kafka_op_type_t type) {
        rd_kafka_op_t *rko;
#define _RD_KAFKA_OP_EMPTY                                                     \
        1234567 /* Special value to be able to assert                          \
                 * on default-initialized (0) sizes                            \
                 * if we forgot to add an op type to                           \
                 * this list. */
        static const size_t op2size[RD_KAFKA_OP__END] = {
            [RD_KAFKA_OP_FETCH]            = sizeof(rko->rko_u.fetch),
            [RD_KAFKA_OP_ERR]              = sizeof(rko->rko_u.err),
            [RD_KAFKA_OP_CONSUMER_ERR]     = sizeof(rko->rko_u.err),
            [RD_KAFKA_OP_DR]               = sizeof(rko->rko_u.dr),
            [RD_KAFKA_OP_STATS]            = sizeof(rko->rko_u.stats),
            [RD_KAFKA_OP_OFFSET_COMMIT]    = sizeof(rko->rko_u.offset_commit),
            [RD_KAFKA_OP_NODE_UPDATE]      = sizeof(rko->rko_u.node),
            [RD_KAFKA_OP_XMIT_BUF]         = sizeof(rko->rko_u.xbuf),
            [RD_KAFKA_OP_RECV_BUF]         = sizeof(rko->rko_u.xbuf),
            [RD_KAFKA_OP_XMIT_RETRY]       = sizeof(rko->rko_u.xbuf),
            [RD_KAFKA_OP_FETCH_START]      = sizeof(rko->rko_u.fetch_start),
            [RD_KAFKA_OP_FETCH_STOP]       = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_SEEK]             = sizeof(rko->rko_u.fetch_start),
            [RD_KAFKA_OP_PAUSE]            = sizeof(rko->rko_u.pause),
            [RD_KAFKA_OP_OFFSET_FETCH]     = sizeof(rko->rko_u.offset_fetch),
            [RD_KAFKA_OP_PARTITION_JOIN]   = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_PARTITION_LEAVE]  = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_REBALANCE]        = sizeof(rko->rko_u.rebalance),
            [RD_KAFKA_OP_TERMINATE]        = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_COORD_QUERY]      = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_SUBSCRIBE]        = sizeof(rko->rko_u.subscribe),
            [RD_KAFKA_OP_ASSIGN]           = sizeof(rko->rko_u.assign),
            [RD_KAFKA_OP_GET_SUBSCRIPTION] = sizeof(rko->rko_u.subscribe),
            [RD_KAFKA_OP_GET_ASSIGNMENT]   = sizeof(rko->rko_u.assign),
            [RD_KAFKA_OP_THROTTLE]         = sizeof(rko->rko_u.throttle),
            [RD_KAFKA_OP_NAME]             = sizeof(rko->rko_u.name),
            [RD_KAFKA_OP_CG_METADATA]      = sizeof(rko->rko_u.cg_metadata),
            [RD_KAFKA_OP_OFFSET_RESET]     = sizeof(rko->rko_u.offset_reset),
            [RD_KAFKA_OP_METADATA]         = sizeof(rko->rko_u.metadata),
            [RD_KAFKA_OP_LOG]              = sizeof(rko->rko_u.log),
            [RD_KAFKA_OP_WAKEUP]           = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_CREATETOPICS]     = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DELETETOPICS]     = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_CREATEPARTITIONS] = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_ALTERCONFIGS]     = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_INCREMENTALALTERCONFIGS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBECONFIGS]    = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DELETERECORDS]      = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_LISTCONSUMERGROUPS] = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBECONSUMERGROUPS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBETOPICS]  = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBECLUSTER] = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DELETEGROUPS]    = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DELETECONSUMERGROUPOFFSETS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_CREATEACLS]   = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBEACLS] = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DELETEACLS]   = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_ALTERCONSUMERGROUPOFFSETS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_LISTCONSUMERGROUPOFFSETS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_ADMIN_FANOUT] = sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_ADMIN_RESULT] = sizeof(rko->rko_u.admin_result),
            [RD_KAFKA_OP_PURGE]        = sizeof(rko->rko_u.purge),
            [RD_KAFKA_OP_CONNECT]      = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_OAUTHBEARER_REFRESH] = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_MOCK]                = sizeof(rko->rko_u.mock),
            [RD_KAFKA_OP_BROKER_MONITOR] = sizeof(rko->rko_u.broker_monitor),
            [RD_KAFKA_OP_TXN]            = sizeof(rko->rko_u.txn),
            [RD_KAFKA_OP_GET_REBALANCE_PROTOCOL] =
                sizeof(rko->rko_u.rebalance_protocol),
            [RD_KAFKA_OP_LEADERS]     = sizeof(rko->rko_u.leaders),
            [RD_KAFKA_OP_BARRIER]     = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_SASL_REAUTH] = _RD_KAFKA_OP_EMPTY,
            [RD_KAFKA_OP_ALTERUSERSCRAMCREDENTIALS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_DESCRIBEUSERSCRAMCREDENTIALS] =
                sizeof(rko->rko_u.admin_request),
            [RD_KAFKA_OP_LISTOFFSETS] = sizeof(rko->rko_u.admin_request),
        };
        size_t tsize = op2size[type & ~RD_KAFKA_OP_FLAGMASK];

        rd_assert(tsize > 0 || !*"add OP type to rd_kafka_op_new0()");
        if (tsize == _RD_KAFKA_OP_EMPTY)
                tsize = 0;

        rko           = rd_calloc(1, sizeof(*rko) - sizeof(rko->rko_u) + tsize);
        rko->rko_type = type;

#if ENABLE_DEVEL
        rko->rko_source = source;
        rd_atomic32_add(&rd_kafka_op_cnt, 1);
#endif
        return rko;
}


void rd_kafka_op_destroy(rd_kafka_op_t *rko) {

        /* Call ops callback with ERR__DESTROY to let it
         * clean up its resources. */
        if ((rko->rko_type & RD_KAFKA_OP_CB) && rko->rko_op_cb) {
                rd_kafka_op_res_t res;
                rko->rko_err = RD_KAFKA_RESP_ERR__DESTROY;
                res          = rko->rko_op_cb(rko->rko_rk, NULL, rko);
                rd_assert(res != RD_KAFKA_OP_RES_YIELD);
                rd_assert(res != RD_KAFKA_OP_RES_KEEP);
        }


        switch (rko->rko_type & ~RD_KAFKA_OP_FLAGMASK) {
        case RD_KAFKA_OP_FETCH:
                rd_kafka_msg_destroy(NULL, &rko->rko_u.fetch.rkm);
                /* Decrease refcount on rkbuf to eventually rd_free shared buf*/
                if (rko->rko_u.fetch.rkbuf)
                        rd_kafka_buf_handle_op(rko, RD_KAFKA_RESP_ERR__DESTROY);

                break;

        case RD_KAFKA_OP_OFFSET_FETCH:
                if (rko->rko_u.offset_fetch.partitions &&
                    rko->rko_u.offset_fetch.do_free)
                        rd_kafka_topic_partition_list_destroy(
                            rko->rko_u.offset_fetch.partitions);
                break;

        case RD_KAFKA_OP_OFFSET_COMMIT:
                RD_IF_FREE(rko->rko_u.offset_commit.partitions,
                           rd_kafka_topic_partition_list_destroy);
                RD_IF_FREE(rko->rko_u.offset_commit.reason, rd_free);
                break;

        case RD_KAFKA_OP_SUBSCRIBE:
        case RD_KAFKA_OP_GET_SUBSCRIPTION:
                RD_IF_FREE(rko->rko_u.subscribe.topics,
                           rd_kafka_topic_partition_list_destroy);
                break;

        case RD_KAFKA_OP_ASSIGN:
        case RD_KAFKA_OP_GET_ASSIGNMENT:
                RD_IF_FREE(rko->rko_u.assign.partitions,
                           rd_kafka_topic_partition_list_destroy);
                break;

        case RD_KAFKA_OP_REBALANCE:
                RD_IF_FREE(rko->rko_u.rebalance.partitions,
                           rd_kafka_topic_partition_list_destroy);
                break;

        case RD_KAFKA_OP_NAME:
                RD_IF_FREE(rko->rko_u.name.str, rd_free);
                break;

        case RD_KAFKA_OP_CG_METADATA:
                RD_IF_FREE(rko->rko_u.cg_metadata,
                           rd_kafka_consumer_group_metadata_destroy);
                break;

        case RD_KAFKA_OP_ERR:
        case RD_KAFKA_OP_CONSUMER_ERR:
                RD_IF_FREE(rko->rko_u.err.errstr, rd_free);
                rd_kafka_msg_destroy(NULL, &rko->rko_u.err.rkm);
                break;

                break;

        case RD_KAFKA_OP_THROTTLE:
                RD_IF_FREE(rko->rko_u.throttle.nodename, rd_free);
                break;

        case RD_KAFKA_OP_STATS:
                RD_IF_FREE(rko->rko_u.stats.json, rd_free);
                break;

        case RD_KAFKA_OP_XMIT_RETRY:
        case RD_KAFKA_OP_XMIT_BUF:
        case RD_KAFKA_OP_RECV_BUF:
                if (rko->rko_u.xbuf.rkbuf)
                        rd_kafka_buf_handle_op(rko, RD_KAFKA_RESP_ERR__DESTROY);

                RD_IF_FREE(rko->rko_u.xbuf.rkbuf, rd_kafka_buf_destroy);
                break;

        case RD_KAFKA_OP_DR:
                rd_kafka_msgq_purge(rko->rko_rk, &rko->rko_u.dr.msgq);
                if (rko->rko_u.dr.do_purge2)
                        rd_kafka_msgq_purge(rko->rko_rk, &rko->rko_u.dr.msgq2);

                if (rko->rko_u.dr.rkt)
                        rd_kafka_topic_destroy0(rko->rko_u.dr.rkt);
                break;

        case RD_KAFKA_OP_OFFSET_RESET:
                RD_IF_FREE(rko->rko_u.offset_reset.reason, rd_free);
                break;

        case RD_KAFKA_OP_METADATA:
                RD_IF_FREE(rko->rko_u.metadata.md, rd_kafka_metadata_destroy);
                /* It's not needed to free metadata.mdi because they
                   are the in the same memory allocation. */
                break;

        case RD_KAFKA_OP_LOG:
                rd_free(rko->rko_u.log.str);
                break;

        case RD_KAFKA_OP_ADMIN_FANOUT:
                rd_assert(rko->rko_u.admin_request.fanout.outstanding == 0);
                rd_list_destroy(&rko->rko_u.admin_request.fanout.results);
        case RD_KAFKA_OP_CREATETOPICS:
        case RD_KAFKA_OP_DELETETOPICS:
        case RD_KAFKA_OP_CREATEPARTITIONS:
        case RD_KAFKA_OP_ALTERCONFIGS:
        case RD_KAFKA_OP_INCREMENTALALTERCONFIGS:
        case RD_KAFKA_OP_DESCRIBECONFIGS:
        case RD_KAFKA_OP_DELETERECORDS:
        case RD_KAFKA_OP_LISTCONSUMERGROUPS:
        case RD_KAFKA_OP_DESCRIBECONSUMERGROUPS:
        case RD_KAFKA_OP_DELETEGROUPS:
        case RD_KAFKA_OP_DELETECONSUMERGROUPOFFSETS:
        case RD_KAFKA_OP_CREATEACLS:
        case RD_KAFKA_OP_DESCRIBEACLS:
        case RD_KAFKA_OP_DELETEACLS:
        case RD_KAFKA_OP_ALTERCONSUMERGROUPOFFSETS:
        case RD_KAFKA_OP_DESCRIBETOPICS:
        case RD_KAFKA_OP_DESCRIBECLUSTER:
        case RD_KAFKA_OP_LISTCONSUMERGROUPOFFSETS:
        case RD_KAFKA_OP_ALTERUSERSCRAMCREDENTIALS:
        case RD_KAFKA_OP_DESCRIBEUSERSCRAMCREDENTIALS:
        case RD_KAFKA_OP_LISTOFFSETS:
                rd_kafka_replyq_destroy(&rko->rko_u.admin_request.replyq);
                rd_list_destroy(&rko->rko_u.admin_request.args);
                if (rko->rko_u.admin_request.options.match_consumer_group_states
                        .u.PTR) {
                        rd_list_destroy(rko->rko_u.admin_request.options
                                            .match_consumer_group_states.u.PTR);
                }
                rd_assert(!rko->rko_u.admin_request.fanout_parent);
                RD_IF_FREE(rko->rko_u.admin_request.coordkey, rd_free);
                break;

        case RD_KAFKA_OP_ADMIN_RESULT:
                rd_list_destroy(&rko->rko_u.admin_result.args);
                rd_list_destroy(&rko->rko_u.admin_result.results);
                RD_IF_FREE(rko->rko_u.admin_result.errstr, rd_free);
                rd_assert(!rko->rko_u.admin_result.fanout_parent);
                ;
                break;

        case RD_KAFKA_OP_MOCK:
                RD_IF_FREE(rko->rko_u.mock.name, rd_free);
                RD_IF_FREE(rko->rko_u.mock.str, rd_free);
                break;

        case RD_KAFKA_OP_BROKER_MONITOR:
                rd_kafka_broker_destroy(rko->rko_u.broker_monitor.rkb);
                break;

        case RD_KAFKA_OP_TXN:
                RD_IF_FREE(rko->rko_u.txn.group_id, rd_free);
                RD_IF_FREE(rko->rko_u.txn.offsets,
                           rd_kafka_topic_partition_list_destroy);
                RD_IF_FREE(rko->rko_u.txn.cgmetadata,
                           rd_kafka_consumer_group_metadata_destroy);
                break;

        case RD_KAFKA_OP_LEADERS:
                rd_assert(!rko->rko_u.leaders.eonce);
                rd_assert(!rko->rko_u.leaders.replyq.q);
                RD_IF_FREE(rko->rko_u.leaders.leaders, rd_list_destroy);
                RD_IF_FREE(rko->rko_u.leaders.partitions,
                           rd_kafka_topic_partition_list_destroy);
                break;

        default:
                break;
        }

        RD_IF_FREE(rko->rko_rktp, rd_kafka_toppar_destroy);

        RD_IF_FREE(rko->rko_error, rd_kafka_error_destroy);

        rd_kafka_replyq_destroy(&rko->rko_replyq);

#if ENABLE_DEVEL
        if (rd_atomic32_sub(&rd_kafka_op_cnt, 1) < 0)
                rd_kafka_assert(NULL, !*"rd_kafka_op_cnt < 0");
#endif

        rd_free(rko);
}



/**
 * Propagate an error event to the application on a specific queue.
 */
void rd_kafka_q_op_err(rd_kafka_q_t *rkq,
                       rd_kafka_resp_err_t err,
                       const char *fmt,
                       ...) {
        va_list ap;
        char buf[2048];
        rd_kafka_op_t *rko;

        va_start(ap, fmt);
        rd_vsnprintf(buf, sizeof(buf), fmt, ap);
        va_end(ap);

        rko                   = rd_kafka_op_new(RD_KAFKA_OP_ERR);
        rko->rko_err          = err;
        rko->rko_u.err.errstr = rd_strdup(buf);

        rd_kafka_q_enq(rkq, rko);
}



/**
 * @brief Enqueue RD_KAFKA_OP_CONSUMER_ERR on \p rkq.
 *
 * @param broker_id Is the relevant broker id, or RD_KAFKA_NODEID_UA (-1)
 *                  if not applicable.
 * @param err Error code.
 * @param version Queue version barrier, or 0 if not applicable.
 * @param topic May be NULL.
 * @param rktp May be NULL. Takes precedence over \p topic.
 * @param offset RD_KAFKA_OFFSET_INVALID if not applicable.
 *
 * @sa rd_kafka_q_op_err()
 */
void rd_kafka_consumer_err(rd_kafka_q_t *rkq,
                           int32_t broker_id,
                           rd_kafka_resp_err_t err,
                           int32_t version,
                           const char *topic,
                           rd_kafka_toppar_t *rktp,
                           int64_t offset,
                           const char *fmt,
                           ...) {
        va_list ap;
        char buf[2048];
        rd_kafka_op_t *rko;

        va_start(ap, fmt);
        rd_vsnprintf(buf, sizeof(buf), fmt, ap);
        va_end(ap);

        rko                   = rd_kafka_op_new(RD_KAFKA_OP_CONSUMER_ERR);
        rko->rko_version      = version;
        rko->rko_err          = err;
        rko->rko_u.err.offset = offset;
        rko->rko_u.err.errstr = rd_strdup(buf);
        rko->rko_u.err.rkm.rkm_broker_id = broker_id;

        if (rktp)
                rko->rko_rktp = rd_kafka_toppar_keep(rktp);
        else if (topic)
                rko->rko_u.err.rkm.rkm_rkmessage.rkt =
                    (rd_kafka_topic_t *)rd_kafka_lwtopic_new(rkq->rkq_rk,
                                                             topic);


        rd_kafka_q_enq(rkq, rko);
}


/**
 * Creates a reply op based on 'rko_orig'.
 * If 'rko_orig' has rko_op_cb set the reply op will be OR:ed with
 * RD_KAFKA_OP_CB, else the reply type will be the original rko_type OR:ed
 * with RD_KAFKA_OP_REPLY.
 */
rd_kafka_op_t *rd_kafka_op_new_reply(rd_kafka_op_t *rko_orig,
                                     rd_kafka_resp_err_t err) {
        rd_kafka_op_t *rko;

        rko = rd_kafka_op_new(rko_orig->rko_type | RD_KAFKA_OP_REPLY);
        rd_kafka_op_get_reply_version(rko, rko_orig);
        rko->rko_err = err;
        if (rko_orig->rko_rktp)
                rko->rko_rktp = rd_kafka_toppar_keep(rko_orig->rko_rktp);

        return rko;
}


/**
 * @brief Create new callback op for type \p type
 */
rd_kafka_op_t *rd_kafka_op_new_cb(rd_kafka_t *rk,
                                  rd_kafka_op_type_t type,
                                  rd_kafka_op_cb_t *cb) {
        rd_kafka_op_t *rko;
        rko            = rd_kafka_op_new(type | RD_KAFKA_OP_CB);
        rko->rko_op_cb = cb;
        rko->rko_rk    = rk;
        return rko;
}


/**
 * @brief Reply to 'rko' re-using the same rko with rko_err
 *        specified by \p err. rko_error is set to NULL.
 *
 * If there is no replyq the rko is destroyed.
 *
 * @returns 1 if op was enqueued, else 0 and rko is destroyed.
 */
int rd_kafka_op_reply(rd_kafka_op_t *rko, rd_kafka_resp_err_t err) {

        if (!rko->rko_replyq.q) {
                rd_kafka_op_destroy(rko);
                return 0;
        }

        rko->rko_type |= (rko->rko_op_cb ? RD_KAFKA_OP_CB : RD_KAFKA_OP_REPLY);
        rko->rko_err   = err;
        rko->rko_error = NULL;

        return rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0);
}


/**
 * @brief Reply to 'rko' re-using the same rko with rko_error specified
 *        by \p error (may be NULL) and rko_err set to the corresponding
 *        error code. Assumes ownership of \p error.
 *
 * If there is no replyq the rko is destroyed.
 *
 * @returns 1 if op was enqueued, else 0 and rko is destroyed.
 */
int rd_kafka_op_error_reply(rd_kafka_op_t *rko, rd_kafka_error_t *error) {

        if (!rko->rko_replyq.q) {
                RD_IF_FREE(error, rd_kafka_error_destroy);
                rd_kafka_op_destroy(rko);
                return 0;
        }

        rko->rko_type |= (rko->rko_op_cb ? RD_KAFKA_OP_CB : RD_KAFKA_OP_REPLY);
        rko->rko_err =
            error ? rd_kafka_error_code(error) : RD_KAFKA_RESP_ERR_NO_ERROR;
        rko->rko_error = error;

        return rd_kafka_replyq_enq(&rko->rko_replyq, rko, 0);
}


/**
 * @brief Send request to queue, wait for response.
 *
 * @returns response on success or NULL if destq is disabled.
 */
rd_kafka_op_t *rd_kafka_op_req0(rd_kafka_q_t *destq,
                                rd_kafka_q_t *recvq,
                                rd_kafka_op_t *rko,
                                int timeout_ms) {
        rd_kafka_op_t *reply;

        /* Indicate to destination where to send reply. */
        rd_kafka_op_set_replyq(rko, recvq, NULL);

        /* Enqueue op */
        if (!rd_kafka_q_enq(destq, rko))
                return NULL;

        /* Wait for reply */
        reply = rd_kafka_q_pop(recvq, rd_timeout_us(timeout_ms), 0);

        /* May be NULL for timeout */
        return reply;
}

/**
 * Send request to queue, wait for response.
 * Creates a temporary reply queue.
 */
rd_kafka_op_t *
rd_kafka_op_req(rd_kafka_q_t *destq, rd_kafka_op_t *rko, int timeout_ms) {
        rd_kafka_q_t *recvq;
        rd_kafka_op_t *reply;

        recvq = rd_kafka_q_new(destq->rkq_rk);

        reply = rd_kafka_op_req0(destq, recvq, rko, timeout_ms);

        rd_kafka_q_destroy_owner(recvq);

        return reply;
}


/**
 * Send simple type-only request to queue, wait for response.
 */
rd_kafka_op_t *rd_kafka_op_req2(rd_kafka_q_t *destq, rd_kafka_op_type_t type) {
        rd_kafka_op_t *rko;

        rko = rd_kafka_op_new(type);
        return rd_kafka_op_req(destq, rko, RD_POLL_INFINITE);
}


/**
 * Destroys the rko and returns its err.
 */
rd_kafka_resp_err_t rd_kafka_op_err_destroy(rd_kafka_op_t *rko) {
        rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR__TIMED_OUT;

        if (rko) {
                err = rko->rko_err;
                rd_kafka_op_destroy(rko);
        }
        return err;
}


/**
 * Destroys the rko and returns its error object or NULL if no error.
 */
rd_kafka_error_t *rd_kafka_op_error_destroy(rd_kafka_op_t *rko) {
        if (rko) {
                rd_kafka_error_t *error = rko->rko_error;
                rko->rko_error          = NULL;
                rd_kafka_op_destroy(rko);
                return error;
        }

        return rd_kafka_error_new(RD_KAFKA_RESP_ERR__TIMED_OUT,
                                  "Operation timed out");
}


/**
 * Call op callback
 */
rd_kafka_op_res_t
rd_kafka_op_call(rd_kafka_t *rk, rd_kafka_q_t *rkq, rd_kafka_op_t *rko) {
        rd_kafka_op_res_t res;
        rd_assert(rko->rko_op_cb);
        res = rko->rko_op_cb(rk, rkq, rko);
        if (unlikely(res == RD_KAFKA_OP_RES_YIELD || rd_kafka_yield_thread))
                return RD_KAFKA_OP_RES_YIELD;
        if (res != RD_KAFKA_OP_RES_KEEP)
                rko->rko_op_cb = NULL;
        return res;
}


/**
 * @brief Creates a new RD_KAFKA_OP_FETCH op representing a
 *        control message. The rkm_flags property is set to
 *        RD_KAFKA_MSG_F_CONTROL.
 */
rd_kafka_op_t *rd_kafka_op_new_ctrl_msg(rd_kafka_toppar_t *rktp,
                                        int32_t version,
                                        rd_kafka_buf_t *rkbuf,
                                        rd_kafka_fetch_pos_t pos) {
        rd_kafka_msg_t *rkm;
        rd_kafka_op_t *rko;

        rko = rd_kafka_op_new_fetch_msg(&rkm, rktp, version, rkbuf, pos, 0,
                                        NULL, 0, NULL);

        rkm->rkm_flags |= RD_KAFKA_MSG_F_CONTROL;

        return rko;
}

/**
 * @brief Creates a new RD_KAFKA_OP_FETCH op and sets up the
 *        embedded message according to the parameters.
 *
 * @param rkmp will be set to the embedded rkm in the rko (for convenience)
 * @param offset may be updated later if relative offset.
 */
rd_kafka_op_t *rd_kafka_op_new_fetch_msg(rd_kafka_msg_t **rkmp,
                                         rd_kafka_toppar_t *rktp,
                                         int32_t version,
                                         rd_kafka_buf_t *rkbuf,
                                         rd_kafka_fetch_pos_t pos,
                                         size_t key_len,
                                         const void *key,
                                         size_t val_len,
                                         const void *val) {
        rd_kafka_msg_t *rkm;
        rd_kafka_op_t *rko;

        rko              = rd_kafka_op_new(RD_KAFKA_OP_FETCH);
        rko->rko_rktp    = rd_kafka_toppar_keep(rktp);
        rko->rko_version = version;
        rkm              = &rko->rko_u.fetch.rkm;
        *rkmp            = rkm;

        /* Since all the ops share the same payload buffer
         * a refcnt is used on the rkbuf that makes sure all
         * consume_cb() will have been
         * called for each of these ops before the rkbuf
         * and its memory backing buffers are freed. */
        rko->rko_u.fetch.rkbuf = rkbuf;
        rd_kafka_buf_keep(rkbuf);

        rkm->rkm_offset                  = pos.offset;
        rkm->rkm_u.consumer.leader_epoch = pos.leader_epoch;

        rkm->rkm_key     = (void *)key;
        rkm->rkm_key_len = key_len;

        rkm->rkm_payload = (void *)val;
        rkm->rkm_len     = val_len;
        rko->rko_len     = (int32_t)rkm->rkm_len;

        rkm->rkm_partition = rktp->rktp_partition;

        /* Persistence status is always PERSISTED for consumed messages
         * since we managed to read the message. */
        rkm->rkm_status = RD_KAFKA_MSG_STATUS_PERSISTED;

        return rko;
}


/**
 * Enqueue ERR__THROTTLE op, if desired.
 */
void rd_kafka_op_throttle_time(rd_kafka_broker_t *rkb,
                               rd_kafka_q_t *rkq,
                               int throttle_time) {
        rd_kafka_op_t *rko;

        if (unlikely(throttle_time > 0))
                rd_avg_add(&rkb->rkb_avg_throttle, throttle_time);

        /* We send throttle events when:
         *  - throttle_time > 0
         *  - throttle_time == 0 and last throttle_time > 0
         */
        if (!rkb->rkb_rk->rk_conf.throttle_cb ||
            (!throttle_time &&
             !rd_atomic32_get(&rkb->rkb_rk->rk_last_throttle)))
                return;

        rd_atomic32_set(&rkb->rkb_rk->rk_last_throttle, throttle_time);

        rko = rd_kafka_op_new(RD_KAFKA_OP_THROTTLE);
        rd_kafka_op_set_prio(rko, RD_KAFKA_PRIO_HIGH);
        rko->rko_u.throttle.nodename      = rd_strdup(rkb->rkb_nodename);
        rko->rko_u.throttle.nodeid        = rkb->rkb_nodeid;
        rko->rko_u.throttle.throttle_time = throttle_time;
        rd_kafka_q_enq(rkq, rko);
}


/**
 * @brief Handle standard op types.
 */
rd_kafka_op_res_t rd_kafka_op_handle_std(rd_kafka_t *rk,
                                         rd_kafka_q_t *rkq,
                                         rd_kafka_op_t *rko,
                                         int cb_type) {
        if (cb_type == RD_KAFKA_Q_CB_FORCE_RETURN)
                return RD_KAFKA_OP_RES_PASS;
        else if (unlikely(rd_kafka_op_is_ctrl_msg(rko))) {
                /* Control messages must not be exposed to the application
                 * but we need to store their offsets. */
                rd_kafka_fetch_op_app_prepare(rk, rko);
                return RD_KAFKA_OP_RES_HANDLED;
        } else if (cb_type != RD_KAFKA_Q_CB_EVENT &&
                   rko->rko_type & RD_KAFKA_OP_CB)
                return rd_kafka_op_call(rk, rkq, rko);
        else if (rko->rko_type == RD_KAFKA_OP_RECV_BUF) /* Handle Response */
                rd_kafka_buf_handle_op(rko, rko->rko_err);
        else if (cb_type != RD_KAFKA_Q_CB_RETURN &&
                 rko->rko_type & RD_KAFKA_OP_REPLY &&
                 rko->rko_err == RD_KAFKA_RESP_ERR__DESTROY)
                return RD_KAFKA_OP_RES_HANDLED; /* dest queue was
                                                 * probably disabled. */
        else
                return RD_KAFKA_OP_RES_PASS;

        return RD_KAFKA_OP_RES_HANDLED;
}


/**
 * @brief Attempt to handle op using its queue's serve callback,
 *        or the passed callback, or op_handle_std(), else do nothing.
 *
 * @param rkq is \p rko's queue (which it was unlinked from) with rkq_lock
 *            being held. Callback may re-enqueue the op on this queue
 *            and return YIELD.
 *
 * @returns HANDLED if op was handled (and destroyed), PASS if not,
 *          or YIELD if op was handled (maybe destroyed or re-enqueued)
 *          and caller must propagate yield upwards (cancel and return).
 */
rd_kafka_op_res_t rd_kafka_op_handle(rd_kafka_t *rk,
                                     rd_kafka_q_t *rkq,
                                     rd_kafka_op_t *rko,
                                     rd_kafka_q_cb_type_t cb_type,
                                     void *opaque,
                                     rd_kafka_q_serve_cb_t *callback) {
        rd_kafka_op_res_t res;

        if (rko->rko_serve) {
                callback              = rko->rko_serve;
                opaque                = rko->rko_serve_opaque;
                rko->rko_serve        = NULL;
                rko->rko_serve_opaque = NULL;
        }

        res = rd_kafka_op_handle_std(rk, rkq, rko, cb_type);
        if (res == RD_KAFKA_OP_RES_KEEP) {
                /* Op was handled but must not be destroyed. */
                return res;
        }
        if (res == RD_KAFKA_OP_RES_HANDLED) {
                rd_kafka_op_destroy(rko);
                return res;
        } else if (unlikely(res == RD_KAFKA_OP_RES_YIELD))
                return res;

        if (callback)
                res = callback(rk, rkq, rko, cb_type, opaque);

        return res;
}


/**
 * @brief Prepare passing message to application.
 *        This must be called just prior to passing/returning a consumed
 *        message to the application.
 *
 * Performs:
 *  - Store offset for fetched message + 1.
 *  - Updates the application offset (rktp_app_offset).
 *
 * @locks rktp_lock and rk_lock MUST NOT be held
 */
void rd_kafka_fetch_op_app_prepare(rd_kafka_t *rk, rd_kafka_op_t *rko) {
        rd_kafka_toppar_t *rktp;
        rd_kafka_fetch_pos_t pos;

        if (unlikely(rko->rko_type != RD_KAFKA_OP_FETCH || rko->rko_err))
                return;

        rktp = rko->rko_rktp;

        if (unlikely(!rk))
                rk = rktp->rktp_rkt->rkt_rk;

        pos.offset       = rko->rko_u.fetch.rkm.rkm_rkmessage.offset + 1;
        pos.leader_epoch = rko->rko_u.fetch.rkm.rkm_u.consumer.leader_epoch;

        rd_kafka_update_app_pos(rk, rktp, pos, RD_DO_LOCK);
}