/*
 * librdkafka - The Apache Kafka C/C++ library
 *
 * Copyright (c) 2015 Magnus Edenhill
 * 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 "rdkafka_int.h"
#include "rdkafka_topic.h"
#include "rdkafka_broker.h"
#include "rdkafka_request.h"
#include "rdkafka_offset.h"
#include "rdkafka_partition.h"
#include "rdregex.h"
#include "rdports.h"  /* rd_qsort_r() */

const char *rd_kafka_fetch_states[] = {
	"none",
        "stopping",
        "stopped",
	"offset-query",
	"offset-wait",
	"active"
};


static rd_kafka_op_res_t
rd_kafka_toppar_op_serve (rd_kafka_t *rk,
                          rd_kafka_q_t *rkq, rd_kafka_op_t *rko,
                          rd_kafka_q_cb_type_t cb_type, void *opaque);


static RD_INLINE int32_t
rd_kafka_toppar_version_new_barrier0 (rd_kafka_toppar_t *rktp,
				     const char *func, int line) {
	int32_t version = rd_atomic32_add(&rktp->rktp_version, 1);
	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BARRIER",
		     "%s [%"PRId32"]: %s:%d: new version barrier v%"PRId32,
		     rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
		     func, line, version);
	return version;
}

#define rd_kafka_toppar_version_new_barrier(rktp) \
	rd_kafka_toppar_version_new_barrier0(rktp, __FUNCTION__, __LINE__)


/**
 * Toppar based OffsetResponse handling.
 * This is used for updating the low water mark for consumer lag.
 */
static void rd_kafka_toppar_lag_handle_Offset (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) {
        shptr_rd_kafka_toppar_t *s_rktp = opaque;
        rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(s_rktp);
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_topic_partition_t *rktpar;

        offsets = rd_kafka_topic_partition_list_new(1);

        /* Parse and return Offset */
        err = rd_kafka_handle_Offset(rkb->rkb_rk, rkb, err,
                                     rkbuf, request, offsets);

        if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS) {
                rd_kafka_topic_partition_list_destroy(offsets);
                return; /* Retrying */
        }

        if (!err && !(rktpar = rd_kafka_topic_partition_list_find(
                              offsets,
                              rktp->rktp_rkt->rkt_topic->str,
                              rktp->rktp_partition)))
                err = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;

        if (!err) {
                rd_kafka_toppar_lock(rktp);
                rktp->rktp_lo_offset = rktpar->offset;
                rd_kafka_toppar_unlock(rktp);
        }

        rd_kafka_topic_partition_list_destroy(offsets);

        rktp->rktp_wait_consumer_lag_resp = 0;

        rd_kafka_toppar_destroy(s_rktp); /* from request.opaque */
}



/**
 * Request information from broker to keep track of consumer lag.
 *
 * Locality: toppar handle thread
 */
static void rd_kafka_toppar_consumer_lag_req (rd_kafka_toppar_t *rktp) {
	rd_kafka_broker_t *rkb;
        rd_kafka_topic_partition_list_t *partitions;

        if (rktp->rktp_wait_consumer_lag_resp)
                return; /* Previous request not finished yet */

        rkb = rd_kafka_toppar_leader(rktp, 1/*proper brokers only*/);
        if (!rkb)
		return;

        rktp->rktp_wait_consumer_lag_resp = 1;

        partitions = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(partitions,
                                          rktp->rktp_rkt->rkt_topic->str,
                                          rktp->rktp_partition)->offset =
                RD_KAFKA_OFFSET_BEGINNING;

        /* Ask for oldest offset. The newest offset is automatically
         * propagated in FetchResponse.HighwaterMark. */
        rd_kafka_OffsetRequest(rkb, partitions, 0,
                               RD_KAFKA_REPLYQ(rktp->rktp_ops, 0),
                               rd_kafka_toppar_lag_handle_Offset,
                               rd_kafka_toppar_keep(rktp));

        rd_kafka_topic_partition_list_destroy(partitions);

        rd_kafka_broker_destroy(rkb); /* from toppar_leader() */
}



/**
 * Request earliest offset to measure consumer lag
 *
 * Locality: toppar handler thread
 */
static void rd_kafka_toppar_consumer_lag_tmr_cb (rd_kafka_timers_t *rkts,
						 void *arg) {
	rd_kafka_toppar_t *rktp = arg;
	rd_kafka_toppar_consumer_lag_req(rktp);
}


/**
 * Add new partition to topic.
 *
 * Locks: rd_kafka_topic_wrlock() must be held.
 * Locks: rd_kafka_wrlock() must be held.
 */
shptr_rd_kafka_toppar_t *rd_kafka_toppar_new0 (rd_kafka_itopic_t *rkt,
					       int32_t partition,
					       const char *func, int line) {
	rd_kafka_toppar_t *rktp;

	rktp = rd_calloc(1, sizeof(*rktp));

	rktp->rktp_partition = partition;
	rktp->rktp_rkt = rkt;
        rktp->rktp_leader_id = -1;
	rktp->rktp_fetch_state = RD_KAFKA_TOPPAR_FETCH_NONE;
        rktp->rktp_fetch_msg_max_bytes
            = rkt->rkt_rk->rk_conf.fetch_msg_max_bytes;
	rktp->rktp_offset_fp = NULL;
        rd_kafka_offset_stats_reset(&rktp->rktp_offsets);
        rd_kafka_offset_stats_reset(&rktp->rktp_offsets_fin);
        rktp->rktp_hi_offset = RD_KAFKA_OFFSET_INVALID;
	rktp->rktp_lo_offset = RD_KAFKA_OFFSET_INVALID;
	rktp->rktp_app_offset = RD_KAFKA_OFFSET_INVALID;
        rktp->rktp_stored_offset = RD_KAFKA_OFFSET_INVALID;
        rktp->rktp_committed_offset = RD_KAFKA_OFFSET_INVALID;
	rd_kafka_msgq_init(&rktp->rktp_msgq);
        rktp->rktp_msgq_wakeup_fd = -1;
	rd_kafka_msgq_init(&rktp->rktp_xmit_msgq);
	mtx_init(&rktp->rktp_lock, mtx_plain);

        rd_refcnt_init(&rktp->rktp_refcnt, 0);
	rktp->rktp_fetchq = rd_kafka_q_new(rkt->rkt_rk);
        rktp->rktp_ops    = rd_kafka_q_new(rkt->rkt_rk);
        rktp->rktp_ops->rkq_serve = rd_kafka_toppar_op_serve;
        rktp->rktp_ops->rkq_opaque = rktp;
        rd_atomic32_init(&rktp->rktp_version, 1);
	rktp->rktp_op_version = rd_atomic32_get(&rktp->rktp_version);

        /* Consumer: If statistics is available we query the oldest offset
         * of each partition.
         * Since the oldest offset only moves on log retention, we cap this
         * value on the low end to a reasonable value to avoid flooding
         * the brokers with OffsetRequests when our statistics interval is low.
         * FIXME: Use a global timer to collect offsets for all partitions */
        if (rktp->rktp_rkt->rkt_rk->rk_conf.stats_interval_ms > 0 &&
            rkt->rkt_rk->rk_type == RD_KAFKA_CONSUMER &&
            rktp->rktp_partition != RD_KAFKA_PARTITION_UA) {
                int intvl = rkt->rkt_rk->rk_conf.stats_interval_ms;
                if (intvl < 10 * 1000 /* 10s */)
                        intvl = 10 * 1000;
		rd_kafka_timer_start(&rkt->rkt_rk->rk_timers,
				     &rktp->rktp_consumer_lag_tmr,
                                     intvl * 1000ll,
				     rd_kafka_toppar_consumer_lag_tmr_cb,
				     rktp);
        }

        rktp->rktp_s_rkt = rd_kafka_topic_keep(rkt);

	rd_kafka_q_fwd_set(rktp->rktp_ops, rkt->rkt_rk->rk_ops);
	rd_kafka_dbg(rkt->rkt_rk, TOPIC, "TOPPARNEW", "NEW %s [%"PRId32"] %p (at %s:%d)",
		     rkt->rkt_topic->str, rktp->rktp_partition, rktp,
		     func, line);

	return rd_kafka_toppar_keep_src(func, line, rktp);
}



/**
 * Removes a toppar from its duties, global lists, etc.
 *
 * Locks: rd_kafka_toppar_lock() MUST be held
 */
static void rd_kafka_toppar_remove (rd_kafka_toppar_t *rktp) {
        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "TOPPARREMOVE",
                     "Removing toppar %s [%"PRId32"] %p",
                     rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
		     rktp);

	rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
			    &rktp->rktp_offset_query_tmr, 1/*lock*/);
	rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
			    &rktp->rktp_consumer_lag_tmr, 1/*lock*/);

	rd_kafka_q_fwd_set(rktp->rktp_ops, NULL);
}


/**
 * Final destructor for partition.
 */
void rd_kafka_toppar_destroy_final (rd_kafka_toppar_t *rktp) {

        rd_kafka_toppar_remove(rktp);

	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "DESTROY",
		     "%s [%"PRId32"]: %p DESTROY_FINAL",
		     rktp->rktp_rkt->rkt_topic->str,
                     rktp->rktp_partition, rktp);

	/* Clear queues */
	rd_kafka_assert(rktp->rktp_rkt->rkt_rk,
			rd_kafka_msgq_len(&rktp->rktp_xmit_msgq) == 0);
	rd_kafka_dr_msgq(rktp->rktp_rkt, &rktp->rktp_msgq,
			 RD_KAFKA_RESP_ERR__DESTROY);
	rd_kafka_q_destroy_owner(rktp->rktp_fetchq);
        rd_kafka_q_destroy_owner(rktp->rktp_ops);

	rd_kafka_replyq_destroy(&rktp->rktp_replyq);

	rd_kafka_topic_destroy0(rktp->rktp_s_rkt);

	mtx_destroy(&rktp->rktp_lock);

        rd_refcnt_destroy(&rktp->rktp_refcnt);

	rd_free(rktp);
}


/**
 * Set toppar fetching state.
 *
 * Locality: broker thread
 * Locks: rd_kafka_toppar_lock() MUST be held.
 */
void rd_kafka_toppar_set_fetch_state (rd_kafka_toppar_t *rktp,
                                      int fetch_state) {
	rd_kafka_assert(NULL,
			thrd_is_current(rktp->rktp_rkt->rkt_rk->rk_thread));

        if ((int)rktp->rktp_fetch_state == fetch_state)
                return;

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "PARTSTATE",
                     "Partition %.*s [%"PRId32"] changed fetch state %s -> %s",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition,
                     rd_kafka_fetch_states[rktp->rktp_fetch_state],
                     rd_kafka_fetch_states[fetch_state]);

        rktp->rktp_fetch_state = fetch_state;

        if (fetch_state == RD_KAFKA_TOPPAR_FETCH_ACTIVE)
                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk,
                             CONSUMER|RD_KAFKA_DBG_TOPIC,
                             "FETCH",
                             "Partition %.*s [%"PRId32"] start fetching "
                             "at offset %s",
                             RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                             rktp->rktp_partition,
                             rd_kafka_offset2str(rktp->rktp_next_offset));
}


/**
 * Returns the appropriate toppar for a given rkt and partition.
 * The returned toppar has increased refcnt and must be unreffed by calling
 *  rd_kafka_toppar_destroy().
 * May return NULL.
 *
 * If 'ua_on_miss' is true the UA (unassigned) toppar is returned if
 * 'partition' was not known locally, else NULL is returned.
 *
 * Locks: Caller must hold rd_kafka_topic_*lock()
 */
shptr_rd_kafka_toppar_t *rd_kafka_toppar_get0 (const char *func, int line,
                                               const rd_kafka_itopic_t *rkt,
                                               int32_t partition,
                                               int ua_on_miss) {
        shptr_rd_kafka_toppar_t *s_rktp;

	if (partition >= 0 && partition < rkt->rkt_partition_cnt)
		s_rktp = rkt->rkt_p[partition];
	else if (partition == RD_KAFKA_PARTITION_UA || ua_on_miss)
		s_rktp = rkt->rkt_ua;
	else
		return NULL;

	if (s_rktp)
                return rd_kafka_toppar_keep_src(func,line,
                                                rd_kafka_toppar_s2i(s_rktp));

	return NULL;
}


/**
 * Same as rd_kafka_toppar_get() but no need for locking and
 * looks up the topic first.
 *
 * Locality: any
 * Locks: none
 */
shptr_rd_kafka_toppar_t *rd_kafka_toppar_get2 (rd_kafka_t *rk,
                                               const char *topic,
                                               int32_t partition,
                                               int ua_on_miss,
                                               int create_on_miss) {
	shptr_rd_kafka_itopic_t *s_rkt;
        rd_kafka_itopic_t *rkt;
        shptr_rd_kafka_toppar_t *s_rktp;

        rd_kafka_wrlock(rk);

        /* Find or create topic */
	if (unlikely(!(s_rkt = rd_kafka_topic_find(rk, topic, 0/*no-lock*/)))) {
                if (!create_on_miss) {
                        rd_kafka_wrunlock(rk);
                        return NULL;
                }
                s_rkt = rd_kafka_topic_new0(rk, topic, NULL,
					    NULL, 0/*no-lock*/);
                if (!s_rkt) {
                        rd_kafka_wrunlock(rk);
                        rd_kafka_log(rk, LOG_ERR, "TOPIC",
                                     "Failed to create local topic \"%s\": %s",
                                     topic, rd_strerror(errno));
                        return NULL;
                }
        }

        rd_kafka_wrunlock(rk);

        rkt = rd_kafka_topic_s2i(s_rkt);

	rd_kafka_topic_wrlock(rkt);
	s_rktp = rd_kafka_toppar_desired_add(rkt, partition);
	rd_kafka_topic_wrunlock(rkt);

        rd_kafka_topic_destroy0(s_rkt);

	return s_rktp;
}


/**
 * Returns a toppar if it is available in the cluster.
 * '*errp' is set to the error-code if lookup fails.
 *
 * Locks: topic_*lock() MUST be held
 */
shptr_rd_kafka_toppar_t *
rd_kafka_toppar_get_avail (const rd_kafka_itopic_t *rkt,
                           int32_t partition, int ua_on_miss,
                           rd_kafka_resp_err_t *errp) {
	shptr_rd_kafka_toppar_t *s_rktp;

        switch (rkt->rkt_state)
        {
        case RD_KAFKA_TOPIC_S_UNKNOWN:
                /* No metadata received from cluster yet.
                 * Put message in UA partition and re-run partitioner when
                 * cluster comes up. */
		partition = RD_KAFKA_PARTITION_UA;
                break;

        case RD_KAFKA_TOPIC_S_NOTEXISTS:
                /* Topic not found in cluster.
                 * Fail message immediately. */
                *errp = RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC;
                return NULL;

        case RD_KAFKA_TOPIC_S_EXISTS:
                /* Topic exists in cluster. */

                /* Topic exists but has no partitions.
                 * This is usually an transient state following the
                 * auto-creation of a topic. */
                if (unlikely(rkt->rkt_partition_cnt == 0)) {
                        partition = RD_KAFKA_PARTITION_UA;
                        break;
                }

                /* Check that partition exists. */
                if (partition >= rkt->rkt_partition_cnt) {
                        *errp = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;
                        return NULL;
                }
                break;

        default:
                rd_kafka_assert(rkt->rkt_rk, !*"NOTREACHED");
                break;
        }

	/* Get new partition */
	s_rktp = rd_kafka_toppar_get(rkt, partition, 0);

	if (unlikely(!s_rktp)) {
		/* Unknown topic or partition */
		if (rkt->rkt_state == RD_KAFKA_TOPIC_S_NOTEXISTS)
			*errp = RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC;
		else
			*errp = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;

		return NULL;
	}

	return s_rktp;
}


/**
 * Looks for partition 'i' in topic 'rkt's desired list.
 *
 * The desired partition list is the list of partitions that are desired
 * (e.g., by the consumer) but not yet seen on a broker.
 * As soon as the partition is seen on a broker the toppar is moved from
 * the desired list and onto the normal rkt_p array.
 * When the partition on the broker goes away a desired partition is put
 * back on the desired list.
 *
 * Locks: rd_kafka_topic_*lock() must be held.
 * Note: 'rktp' refcount is increased.
 */

shptr_rd_kafka_toppar_t *rd_kafka_toppar_desired_get (rd_kafka_itopic_t *rkt,
                                                      int32_t partition) {
	shptr_rd_kafka_toppar_t *s_rktp;
        int i;

	RD_LIST_FOREACH(s_rktp, &rkt->rkt_desp, i) {
                rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(s_rktp);
		if (rktp->rktp_partition == partition)
			return rd_kafka_toppar_keep(rktp);
        }

	return NULL;
}


/**
 * Link toppar on desired list.
 *
 * Locks: rd_kafka_topic_wrlock() and toppar_lock() must be held.
 */
void rd_kafka_toppar_desired_link (rd_kafka_toppar_t *rktp) {
        shptr_rd_kafka_toppar_t *s_rktp;

        if (rktp->rktp_s_for_desp)
                return; /* Already linked */

        s_rktp = rd_kafka_toppar_keep(rktp);
        rd_list_add(&rktp->rktp_rkt->rkt_desp, s_rktp);
        rktp->rktp_s_for_desp = s_rktp; /* Desired list refcount */
}

/**
 * Unlink toppar from desired list.
 *
 * Locks: rd_kafka_topic_wrlock() and toppar_lock() must be held.
 */
void rd_kafka_toppar_desired_unlink (rd_kafka_toppar_t *rktp) {
        if (!rktp->rktp_s_for_desp)
                return; /* Not linked */

        rd_list_remove(&rktp->rktp_rkt->rkt_desp, rktp->rktp_s_for_desp);
        rd_kafka_toppar_destroy(rktp->rktp_s_for_desp);
        rktp->rktp_s_for_desp = NULL;
 }


/**
 * @brief If rktp is not already desired:
 *  - mark as DESIRED|UNKNOWN
 *  - add to desired list
 *
 * @remark toppar_lock() MUST be held
 */
void rd_kafka_toppar_desired_add0 (rd_kafka_toppar_t *rktp) {
        if ((rktp->rktp_flags & RD_KAFKA_TOPPAR_F_DESIRED))
                return;

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "DESIRED",
                     "%s [%"PRId32"]: adding to DESIRED list",
                     rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition);
	rktp->rktp_flags |= RD_KAFKA_TOPPAR_F_DESIRED;
        rd_kafka_toppar_desired_link(rktp);
}


/**
 * Adds 'partition' as a desired partition to topic 'rkt', or updates
 * an existing partition to be desired.
 *
 * Locks: rd_kafka_topic_wrlock() must be held.
 */
shptr_rd_kafka_toppar_t *rd_kafka_toppar_desired_add (rd_kafka_itopic_t *rkt,
                                                      int32_t partition) {
	shptr_rd_kafka_toppar_t *s_rktp;
        rd_kafka_toppar_t *rktp;

	if ((s_rktp = rd_kafka_toppar_get(rkt,
                                          partition, 0/*no_ua_on_miss*/))) {
                rktp = rd_kafka_toppar_s2i(s_rktp);
		rd_kafka_toppar_lock(rktp);
                if (unlikely(!(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_DESIRED))) {
                        rd_kafka_dbg(rkt->rkt_rk, TOPIC, "DESP",
                                     "Setting topic %s [%"PRId32"] partition "
                                     "as desired",
                                     rkt->rkt_topic->str, rktp->rktp_partition);
                        rktp->rktp_flags |= RD_KAFKA_TOPPAR_F_DESIRED;
                }
		rd_kafka_toppar_unlock(rktp);
		return s_rktp;
	}

	if ((s_rktp = rd_kafka_toppar_desired_get(rkt, partition)))
		return s_rktp;

	s_rktp = rd_kafka_toppar_new(rkt, partition);
        rktp = rd_kafka_toppar_s2i(s_rktp);

        rd_kafka_toppar_lock(rktp);
        rktp->rktp_flags |= RD_KAFKA_TOPPAR_F_UNKNOWN;
        rd_kafka_toppar_desired_add0(rktp);
        rd_kafka_toppar_unlock(rktp);

	rd_kafka_dbg(rkt->rkt_rk, TOPIC, "DESP",
		     "Adding desired topic %s [%"PRId32"]",
		     rkt->rkt_topic->str, rktp->rktp_partition);

	return s_rktp; /* Callers refcount */
}




/**
 * Unmarks an 'rktp' as desired.
 *
 * Locks: rd_kafka_topic_wrlock() and rd_kafka_toppar_lock() MUST be held.
 */
void rd_kafka_toppar_desired_del (rd_kafka_toppar_t *rktp) {

	if (!(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_DESIRED))
		return;

	rktp->rktp_flags &= ~RD_KAFKA_TOPPAR_F_DESIRED;
        rd_kafka_toppar_desired_unlink(rktp);

        if (rktp->rktp_flags & RD_KAFKA_TOPPAR_F_UNKNOWN)
                rktp->rktp_flags &= ~RD_KAFKA_TOPPAR_F_UNKNOWN;


	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "DESP",
		     "Removing (un)desired topic %s [%"PRId32"]",
		     rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition);
}



/**
 * Append message at tail of 'rktp' message queue.
 */
void rd_kafka_toppar_enq_msg (rd_kafka_toppar_t *rktp, rd_kafka_msg_t *rkm) {
        int wakeup_fd, queue_len;

        rd_kafka_toppar_lock(rktp);

        if (!rkm->rkm_u.producer.msgseq &&
            rktp->rktp_partition != RD_KAFKA_PARTITION_UA)
                rkm->rkm_u.producer.msgseq = ++rktp->rktp_msgseq;

        if (rktp->rktp_partition == RD_KAFKA_PARTITION_UA ||
            rktp->rktp_rkt->rkt_conf.queuing_strategy == RD_KAFKA_QUEUE_FIFO) {
                /* No need for enq_sorted(), this is the oldest message. */
                queue_len = rd_kafka_msgq_enq(&rktp->rktp_msgq, rkm);
        } else {
                queue_len = rd_kafka_msgq_enq_sorted(rktp->rktp_rkt,
                                                     &rktp->rktp_msgq, rkm);
        }

        wakeup_fd = rktp->rktp_msgq_wakeup_fd;
        rd_kafka_toppar_unlock(rktp);

#ifndef _MSC_VER
        if (wakeup_fd != -1 && queue_len == 1) {
                char one = 1;
                int r;
                r = rd_write(wakeup_fd, &one, sizeof(one));
                if (r == -1)
                        rd_kafka_log(rktp->rktp_rkt->rkt_rk, LOG_ERR, "PARTENQ",
                                     "%s [%"PRId32"]: write to "
                                     "wake-up fd %d failed: %s",
                                     rktp->rktp_rkt->rkt_topic->str,
                                     rktp->rktp_partition,
                                     wakeup_fd,
                                     rd_strerror(errno));
        }
#endif
}


/**
 * Dequeue message from 'rktp' message queue.
 */
void rd_kafka_toppar_deq_msg (rd_kafka_toppar_t *rktp, rd_kafka_msg_t *rkm) {
	rd_kafka_toppar_lock(rktp);
	rd_kafka_msgq_deq(&rktp->rktp_msgq, rkm, 1);
	rd_kafka_toppar_unlock(rktp);
}


void rd_kafka_msgq_insert_msgq (rd_kafka_msgq_t *destq,
                                rd_kafka_msgq_t *srcq,
                                int (*cmp) (const void *a, const void *b)) {
        rd_kafka_msg_t *first, *dest_first;

        first = TAILQ_FIRST(&srcq->rkmq_msgs);
        if (unlikely(!first)) {
                /* srcq is empty */
                return;
        }

        dest_first = TAILQ_FIRST(&destq->rkmq_msgs);

        /*
         * Try to optimize insertion of source list.
         */

        if (unlikely(!dest_first)) {
                /* Dest queue is empty, simply move the srcq. */
                rd_kafka_msgq_move(destq, srcq);

                return;
        }

        /* See if we can optimize the insertion by bulk-loading
         * the messages in place.
         * We know that:
         *  - destq is sorted
         *  - srcq is sorted
         *  - there is no overlap between the two.
         */

        if (cmp(first, dest_first) < 0) {
                /* Prepend src to dest queue.
                 * First append existing dest queue to src queue,
                 * then move src queue to now-empty dest queue,
                 * effectively prepending src queue to dest queue. */
                rd_kafka_msgq_concat(srcq, destq);
                rd_kafka_msgq_move(destq, srcq);

        } else if (cmp(first,
                       TAILQ_LAST(&destq->rkmq_msgs,
                                  rd_kafka_msgs_head_s)) > 0) {
                /* Append src to dest queue */
                rd_kafka_msgq_concat(destq, srcq);

        } else {
                /* Source queue messages reside somewhere
                 * in the dest queue range, find the insert position. */
                rd_kafka_msg_t *at;

                at = rd_kafka_msgq_find_pos(destq, first, cmp);
                rd_assert(at &&
                          *"Bug in msg_order_cmp(): "
                          "could not find insert position");

                /* Insert input queue after 'at' position.
                 * We know that:
                 * - at is non-NULL
                 * - at is not the last element. */
                TAILQ_INSERT_LIST(&destq->rkmq_msgs,
                                  at, &srcq->rkmq_msgs,
                                  rd_kafka_msgs_head_s,
                                  rd_kafka_msg_t *, rkm_link);

                destq->rkmq_msg_cnt   += srcq->rkmq_msg_cnt;
                destq->rkmq_msg_bytes += srcq->rkmq_msg_bytes;
                rd_kafka_msgq_init(srcq);
        }
}


/**
 * @brief Inserts messages from \p srcq according to their sorted position
 *        into \p destq, filtering out messages that can not be retried.
 *
 * @param incr_retry Increment retry count for messages.
 * @param max_retries Maximum retries allowed per message.
 * @param backoff Absolute retry backoff for retried messages.
 *
 * @returns the number of messages that could not be retried.
 */
int rd_kafka_retry_msgq (rd_kafka_msgq_t *destq,
                         rd_kafka_msgq_t *srcq,
                         int incr_retry, int max_retries, rd_ts_t backoff,
                         int (*cmp) (const void *a, const void *b)) {
        rd_kafka_msgq_t retryable = RD_KAFKA_MSGQ_INITIALIZER(retryable);
        rd_kafka_msg_t *rkm, *tmp;

        /* Scan through messages to see which ones are eligible for retry,
         * move the retryable ones to temporary queue and
         * set backoff time for first message and optionally
         * increase retry count for each message.
         * Sorted insert is not necessary since the original order
         * srcq order is maintained. */
        TAILQ_FOREACH_SAFE(rkm, &srcq->rkmq_msgs, rkm_link, tmp) {
                if (rkm->rkm_u.producer.retries + incr_retry > max_retries)
                        continue;

                rd_kafka_msgq_deq(srcq, rkm, 1);
                rd_kafka_msgq_enq(&retryable, rkm);

                rkm->rkm_u.producer.ts_backoff = backoff;
                rkm->rkm_u.producer.retries  += incr_retry;
        }

        /* No messages are retryable */
        if (RD_KAFKA_MSGQ_EMPTY(&retryable))
                return 0;

        /* Insert retryable list at sorted position */
        rd_kafka_msgq_insert_msgq(destq, &retryable, cmp);

        return 1;
}

/**
 * @brief Inserts messages from \p rkmq according to their sorted position
 *        into the partition xmit queue (i.e., the broker xmit work queue).
 *
 * @param incr_retry Increment retry count for messages.
 *
 * @returns the number of messages that could not be retried.
 *
 * @locality Broker thread
 */

int rd_kafka_toppar_retry_msgq (rd_kafka_toppar_t *rktp, rd_kafka_msgq_t *rkmq,
                                int incr_retry) {
        rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
        rd_ts_t backoff = rd_clock() + (rk->rk_conf.retry_backoff_ms * 1000);
        int r;

        rd_kafka_toppar_lock(rktp);
        r = rd_kafka_retry_msgq(&rktp->rktp_xmit_msgq, rkmq,
                                incr_retry, rk->rk_conf.max_retries,
                                backoff,
                                rktp->rktp_rkt->rkt_conf.msg_order_cmp);
        rd_kafka_toppar_unlock(rktp);

        return r;
}

/**
 * @brief Insert sorted message list \p rkmq at sorted position in \p rktp 's
 *        message queue. The queues must not overlap.
 * @remark \p rkmq will be cleared.
 */
void rd_kafka_toppar_insert_msgq (rd_kafka_toppar_t *rktp,
                                  rd_kafka_msgq_t *rkmq) {
        rd_kafka_toppar_lock(rktp);
        rd_kafka_msgq_insert_msgq(&rktp->rktp_msgq, rkmq,
                                  rktp->rktp_rkt->rkt_conf.msg_order_cmp);
        rd_kafka_toppar_unlock(rktp);
}



/**
 * Helper method for purging queues when removing a toppar.
 * Locks: rd_kafka_toppar_lock() MUST be held
 */
void rd_kafka_toppar_purge_queues (rd_kafka_toppar_t *rktp) {
        rd_kafka_q_disable(rktp->rktp_fetchq);
        rd_kafka_q_purge(rktp->rktp_fetchq);
        rd_kafka_q_disable(rktp->rktp_ops);
        rd_kafka_q_purge(rktp->rktp_ops);
}


/**
 * Migrate rktp from (optional) \p old_rkb to (optional) \p new_rkb.
 * This is an async operation.
 *
 * Locks: rd_kafka_toppar_lock() MUST be held
 */
static void rd_kafka_toppar_broker_migrate (rd_kafka_toppar_t *rktp,
                                            rd_kafka_broker_t *old_rkb,
                                            rd_kafka_broker_t *new_rkb) {
        rd_kafka_op_t *rko;
        rd_kafka_broker_t *dest_rkb;
        int had_next_leader = rktp->rktp_next_leader ? 1 : 0;

        /* Update next leader */
        if (new_rkb)
                rd_kafka_broker_keep(new_rkb);
        if (rktp->rktp_next_leader)
                rd_kafka_broker_destroy(rktp->rktp_next_leader);
        rktp->rktp_next_leader = new_rkb;

        /* If next_leader is set it means there is already an async
         * migration op going on and we should not send a new one
         * but simply change the next_leader (which we did above). */
        if (had_next_leader)
                return;

	/* Revert from offset-wait state back to offset-query
	 * prior to leaving the broker to avoid stalling
	 * on the new broker waiting for a offset reply from
	 * this old broker (that might not come and thus need
	 * to time out..slowly) */
	if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT) {
		rd_kafka_toppar_set_fetch_state(
			rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);
		rd_kafka_timer_start(&rktp->rktp_rkt->rkt_rk->rk_timers,
				     &rktp->rktp_offset_query_tmr,
				     500*1000,
				     rd_kafka_offset_query_tmr_cb,
				     rktp);
	}

        if (old_rkb) {
                /* If there is an existing broker for this toppar we let it
                 * first handle its own leave and then trigger the join for
                 * the next leader, if any. */
                rko = rd_kafka_op_new(RD_KAFKA_OP_PARTITION_LEAVE);
                dest_rkb = old_rkb;
        } else {
                /* No existing broker, send join op directly to new leader. */
                rko = rd_kafka_op_new(RD_KAFKA_OP_PARTITION_JOIN);
                dest_rkb = new_rkb;
        }

        rko->rko_rktp = rd_kafka_toppar_keep(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKMIGR",
                     "Migrating topic %.*s [%"PRId32"] %p from %s to %s "
		     "(sending %s to %s)",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition, rktp,
                     old_rkb ? rd_kafka_broker_name(old_rkb) : "(none)",
                     new_rkb ? rd_kafka_broker_name(new_rkb) : "(none)",
		     rd_kafka_op2str(rko->rko_type),
		     rd_kafka_broker_name(dest_rkb));

        rd_kafka_q_enq(dest_rkb->rkb_ops, rko);
}


/**
 * Async toppar leave from broker.
 * Only use this when partitions are to be removed.
 *
 * Locks: rd_kafka_toppar_lock() MUST be held
 */
void rd_kafka_toppar_broker_leave_for_remove (rd_kafka_toppar_t *rktp) {
        rd_kafka_op_t *rko;
        rd_kafka_broker_t *dest_rkb;


	if (rktp->rktp_next_leader)
		dest_rkb = rktp->rktp_next_leader;
	else if (rktp->rktp_leader)
		dest_rkb = rktp->rktp_leader;
	else {
		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "TOPPARDEL",
			     "%.*s [%"PRId32"] %p not handled by any broker: "
			     "not sending LEAVE for remove",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition, rktp);
		return;
	}


	/* Revert from offset-wait state back to offset-query
	 * prior to leaving the broker to avoid stalling
	 * on the new broker waiting for a offset reply from
	 * this old broker (that might not come and thus need
	 * to time out..slowly) */
	if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT)
		rd_kafka_toppar_set_fetch_state(
			rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);

	rko = rd_kafka_op_new(RD_KAFKA_OP_PARTITION_LEAVE);
        rko->rko_rktp = rd_kafka_toppar_keep(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKMIGR",
                     "%.*s [%"PRId32"] %p sending final LEAVE for removal by %s",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition, rktp,
                     rd_kafka_broker_name(dest_rkb));

        rd_kafka_q_enq(dest_rkb->rkb_ops, rko);
}



/**
 * Delegates broker 'rkb' as leader for toppar 'rktp'.
 * 'rkb' may be NULL to undelegate leader.
 *
 * Locks: Caller must have rd_kafka_topic_wrlock(rktp->rktp_rkt) 
 *        AND rd_kafka_toppar_lock(rktp) held.
 */
void rd_kafka_toppar_broker_delegate (rd_kafka_toppar_t *rktp,
				      rd_kafka_broker_t *rkb,
				      int for_removal) {
        rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
        int internal_fallback = 0;

	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
		     "%s [%"PRId32"]: delegate to broker %s "
		     "(rktp %p, term %d, ref %d, remove %d)",
		     rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition,
		     rkb ? rkb->rkb_name : "(none)",
		     rktp, rd_kafka_terminating(rk),
		     rd_refcnt_get(&rktp->rktp_refcnt),
		     for_removal);

        /* Delegate toppars with no leader to the
         * internal broker for bookkeeping. */
        if (!rkb && !for_removal && !rd_kafka_terminating(rk)) {
                rkb = rd_kafka_broker_internal(rk);
                internal_fallback = 1;
        }

	if (rktp->rktp_leader == rkb && !rktp->rktp_next_leader) {
                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
			     "%.*s [%"PRId32"]: not updating broker: "
                             "already on correct broker %s",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition,
                             rkb ? rd_kafka_broker_name(rkb) : "(none)");

                if (internal_fallback)
                        rd_kafka_broker_destroy(rkb);
		return;
        }

	if (rktp->rktp_leader)
		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
			     "%.*s [%"PRId32"]: broker %s no longer leader",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition,
			     rd_kafka_broker_name(rktp->rktp_leader));


	if (rkb) {
		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
			     "%.*s [%"PRId32"]: broker %s is now leader "
			     "for partition with %i messages "
			     "(%"PRIu64" bytes) queued",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition,
			     rd_kafka_broker_name(rkb),
                             rktp->rktp_msgq.rkmq_msg_cnt,
                             rktp->rktp_msgq.rkmq_msg_bytes);


	} else {
		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
			     "%.*s [%"PRId32"]: no leader broker",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition);
	}

        if (rktp->rktp_leader || rkb)
                rd_kafka_toppar_broker_migrate(rktp, rktp->rktp_leader, rkb);

        if (internal_fallback)
                rd_kafka_broker_destroy(rkb);
}





void
rd_kafka_toppar_offset_commit_result (rd_kafka_toppar_t *rktp,
				      rd_kafka_resp_err_t err,
				      rd_kafka_topic_partition_list_t *offsets){
	if (err) {
		rd_kafka_q_op_err(rktp->rktp_fetchq,
				  RD_KAFKA_OP_CONSUMER_ERR,
				  err, 0 /* FIXME:VERSION*/,
				  rktp, 0,
				  "Offset commit failed: %s",
				  rd_kafka_err2str(err));
		return;
	}

	rd_kafka_toppar_lock(rktp);
	rktp->rktp_committed_offset = offsets->elems[0].offset;

	/* When stopping toppars:
	 * Final commit is now done (or failed), propagate. */
	if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_STOPPING)
		rd_kafka_toppar_fetch_stopped(rktp, err);

	rd_kafka_toppar_unlock(rktp);
}


/**
 * Commit toppar's offset on broker.
 * This is an asynch operation, this function simply enqueues an op
 * on the cgrp's queue.
 *
 * Locality: rktp's broker thread
 */
void rd_kafka_toppar_offset_commit (rd_kafka_toppar_t *rktp, int64_t offset,
				    const char *metadata) {
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_topic_partition_t *rktpar;

        rd_kafka_assert(rktp->rktp_rkt->rkt_rk, rktp->rktp_cgrp != NULL);
        rd_kafka_assert(rktp->rktp_rkt->rkt_rk,
                        rktp->rktp_flags & RD_KAFKA_TOPPAR_F_OFFSET_STORE);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, CGRP, "OFFSETCMT",
                     "%.*s [%"PRId32"]: committing offset %"PRId64,
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition, offset);

        offsets = rd_kafka_topic_partition_list_new(1);
        rktpar = rd_kafka_topic_partition_list_add(
                offsets, rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition);
        rktpar->offset = offset;
        if (metadata) {
                rktpar->metadata = rd_strdup(metadata);
                rktpar->metadata_size = strlen(metadata);
        }

        rktp->rktp_committing_offset = offset;

        rd_kafka_commit(rktp->rktp_rkt->rkt_rk, offsets, 1/*async*/);

        rd_kafka_topic_partition_list_destroy(offsets);
}














/**
 * Handle the next offset to consume for a toppar.
 * This is used during initial setup when trying to figure out what
 * offset to start consuming from.
 *
 * Locality: toppar handler thread.
 * Locks: toppar_lock(rktp) must be held
 */
void rd_kafka_toppar_next_offset_handle (rd_kafka_toppar_t *rktp,
                                         int64_t Offset) {

        if (RD_KAFKA_OFFSET_IS_LOGICAL(Offset)) {
                /* Offset storage returned logical offset (e.g. "end"),
                 * look it up. */
                rd_kafka_offset_reset(rktp, Offset, RD_KAFKA_RESP_ERR_NO_ERROR,
                                      "update");
                return;
        }

        /* Adjust by TAIL count if, if wanted */
        if (rktp->rktp_query_offset <=
            RD_KAFKA_OFFSET_TAIL_BASE) {
                int64_t orig_Offset = Offset;
                int64_t tail_cnt =
                        llabs(rktp->rktp_query_offset -
                              RD_KAFKA_OFFSET_TAIL_BASE);

                if (tail_cnt > Offset)
                        Offset = 0;
                else
                        Offset -= tail_cnt;

                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "OFFSET",
                             "OffsetReply for topic %s [%"PRId32"]: "
                             "offset %"PRId64": adjusting for "
                             "OFFSET_TAIL(%"PRId64"): "
                             "effective offset %"PRId64,
                             rktp->rktp_rkt->rkt_topic->str,
                             rktp->rktp_partition,
                             orig_Offset, tail_cnt,
                             Offset);
        }

        rktp->rktp_next_offset = Offset;

        rd_kafka_toppar_set_fetch_state(rktp, RD_KAFKA_TOPPAR_FETCH_ACTIVE);

        /* Wake-up broker thread which might be idling on IO */
        if (rktp->rktp_leader)
                rd_kafka_broker_wakeup(rktp->rktp_leader);

}



/**
 * Fetch stored offset for a single partition. (simple consumer)
 *
 * Locality: toppar thread
 */
void rd_kafka_toppar_offset_fetch (rd_kafka_toppar_t *rktp,
                                   rd_kafka_replyq_t replyq) {
        rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
        rd_kafka_topic_partition_list_t *part;
        rd_kafka_op_t *rko;

        rd_kafka_dbg(rk, TOPIC, "OFFSETREQ",
                     "Partition %.*s [%"PRId32"]: querying cgrp for "
                     "stored offset (opv %d)",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition, replyq.version);

        part = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add0(part,
                                           rktp->rktp_rkt->rkt_topic->str,
                                           rktp->rktp_partition,
					   rd_kafka_toppar_keep(rktp));

        rko = rd_kafka_op_new(RD_KAFKA_OP_OFFSET_FETCH);
	rko->rko_rktp = rd_kafka_toppar_keep(rktp);
	rko->rko_replyq = replyq;

	rko->rko_u.offset_fetch.partitions = part;
	rko->rko_u.offset_fetch.do_free = 1;

        rd_kafka_q_enq(rktp->rktp_cgrp->rkcg_ops, rko);
}




/**
 * Toppar based OffsetResponse handling.
 * This is used for finding the next offset to Fetch.
 *
 * Locality: toppar handler thread
 */
static void rd_kafka_toppar_handle_Offset (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) {
        shptr_rd_kafka_toppar_t *s_rktp = opaque;
        rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(s_rktp);
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_topic_partition_t *rktpar;
        int64_t Offset;

	rd_kafka_toppar_lock(rktp);
	/* Drop reply from previous partition leader */
	if (err != RD_KAFKA_RESP_ERR__DESTROY && rktp->rktp_leader != rkb)
		err = RD_KAFKA_RESP_ERR__OUTDATED;
	rd_kafka_toppar_unlock(rktp);

        offsets = rd_kafka_topic_partition_list_new(1);

        /* Parse and return Offset */
        err = rd_kafka_handle_Offset(rkb->rkb_rk, rkb, err,
                                     rkbuf, request, offsets);

	rd_rkb_dbg(rkb, TOPIC, "OFFSET",
		   "Offset reply for "
		   "topic %.*s [%"PRId32"] (v%d vs v%d)",
		   RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
		   rktp->rktp_partition, request->rkbuf_replyq.version,
		   rktp->rktp_op_version);

	rd_dassert(request->rkbuf_replyq.version > 0);
	if (err != RD_KAFKA_RESP_ERR__DESTROY &&
            rd_kafka_buf_version_outdated(request, rktp->rktp_op_version)) {
		/* Outdated request response, ignore. */
		    err = RD_KAFKA_RESP_ERR__OUTDATED;
	}

        if (!err &&
            (!(rktpar = rd_kafka_topic_partition_list_find(
                       offsets,
                       rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition))))
                err = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;

        if (err) {
                rd_kafka_op_t *rko;

                rd_rkb_dbg(rkb, TOPIC, "OFFSET",
                           "Offset reply error for "
                           "topic %.*s [%"PRId32"] (v%d): %s",
                           RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                           rktp->rktp_partition, request->rkbuf_replyq.version,
			   rd_kafka_err2str(err));

                rd_kafka_topic_partition_list_destroy(offsets);

                if (err == RD_KAFKA_RESP_ERR__DESTROY ||
                    err == RD_KAFKA_RESP_ERR__OUTDATED) {
                        /* Termination or outdated, quick cleanup. */

                        /* from request.opaque */
                        rd_kafka_toppar_destroy(s_rktp);
                        return;

		} else if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
			return; /* Retry in progress */


                rd_kafka_toppar_lock(rktp);
                rd_kafka_offset_reset(rktp, rktp->rktp_query_offset,
                                      err,
                                      "failed to query logical offset");

                /* Signal error back to application,
                 * unless this is an intermittent problem
                 * (e.g.,connection lost) */
                rko = rd_kafka_op_new(RD_KAFKA_OP_CONSUMER_ERR);
                rko->rko_err = err;
                if (rktp->rktp_query_offset <=
                    RD_KAFKA_OFFSET_TAIL_BASE)
                        rko->rko_u.err.offset =
                                rktp->rktp_query_offset -
                                RD_KAFKA_OFFSET_TAIL_BASE;
                else
                        rko->rko_u.err.offset = rktp->rktp_query_offset;
                rd_kafka_toppar_unlock(rktp);
                rko->rko_rktp = rd_kafka_toppar_keep(rktp);

                rd_kafka_q_enq(rktp->rktp_fetchq, rko);

                rd_kafka_toppar_destroy(s_rktp); /* from request.opaque */
                return;
        }

        Offset = rktpar->offset;
        rd_kafka_topic_partition_list_destroy(offsets);

	rd_kafka_toppar_lock(rktp);
        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "OFFSET",
                     "Offset %s request for %.*s [%"PRId32"] "
                     "returned offset %s (%"PRId64")",
                     rd_kafka_offset2str(rktp->rktp_query_offset),
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition, rd_kafka_offset2str(Offset), Offset);

        rd_kafka_toppar_next_offset_handle(rktp, Offset);
	rd_kafka_toppar_unlock(rktp);

        rd_kafka_toppar_destroy(s_rktp); /* from request.opaque */
}

/**
 * Send OffsetRequest for toppar.
 *
 * If \p backoff_ms is non-zero only the query timer is started,
 * otherwise a query is triggered directly.
 *
 * Locality: toppar handler thread
 * Locks: toppar_lock() must be held
 */
void rd_kafka_toppar_offset_request (rd_kafka_toppar_t *rktp,
				     int64_t query_offset, int backoff_ms) {
	rd_kafka_broker_t *rkb;

	rd_kafka_assert(NULL,
			thrd_is_current(rktp->rktp_rkt->rkt_rk->rk_thread));

        rkb = rktp->rktp_leader;

        if (!backoff_ms && (!rkb || rkb->rkb_source == RD_KAFKA_INTERNAL))
                backoff_ms = 500;

        if (backoff_ms) {
		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "OFFSET",
			     "%s [%"PRId32"]: %s"
			     "starting offset query timer for offset %s",
			     rktp->rktp_rkt->rkt_topic->str,
			     rktp->rktp_partition,
                             !rkb ? "no current leader for partition, " : "",
			     rd_kafka_offset2str(query_offset));

                rd_kafka_toppar_set_fetch_state(
                        rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);
		rd_kafka_timer_start(&rktp->rktp_rkt->rkt_rk->rk_timers,
				     &rktp->rktp_offset_query_tmr,
				     backoff_ms*1000ll,
				     rd_kafka_offset_query_tmr_cb, rktp);
		return;
        }


        rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
                            &rktp->rktp_offset_query_tmr, 1/*lock*/);


	if (query_offset == RD_KAFKA_OFFSET_STORED &&
            rktp->rktp_rkt->rkt_conf.offset_store_method ==
            RD_KAFKA_OFFSET_METHOD_BROKER) {
                /*
                 * Get stored offset from broker based storage:
                 * ask cgrp manager for offsets
                 */
                rd_kafka_toppar_offset_fetch(
			rktp,
			RD_KAFKA_REPLYQ(rktp->rktp_ops,
					rktp->rktp_op_version));

	} else {
                shptr_rd_kafka_toppar_t *s_rktp;
                rd_kafka_topic_partition_list_t *offsets;

                /*
                 * Look up logical offset (end,beginning,tail,..)
                 */

                rd_rkb_dbg(rkb, TOPIC, "OFFREQ",
                           "Partition %.*s [%"PRId32"]: querying for logical "
                           "offset %s (opv %d)",
                           RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                           rktp->rktp_partition,
                           rd_kafka_offset2str(query_offset),
			   rktp->rktp_op_version);

                s_rktp = rd_kafka_toppar_keep(rktp);

		if (query_offset <= RD_KAFKA_OFFSET_TAIL_BASE)
			query_offset = RD_KAFKA_OFFSET_END;

                offsets = rd_kafka_topic_partition_list_new(1);
                rd_kafka_topic_partition_list_add(
                        offsets,
                        rktp->rktp_rkt->rkt_topic->str,
                        rktp->rktp_partition)->offset = query_offset;

                rd_kafka_OffsetRequest(rkb, offsets, 0,
                                       RD_KAFKA_REPLYQ(rktp->rktp_ops,
                                                       rktp->rktp_op_version),
                                       rd_kafka_toppar_handle_Offset,
                                       s_rktp);

                rd_kafka_topic_partition_list_destroy(offsets);
        }

        rd_kafka_toppar_set_fetch_state(rktp,
					RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT);
}


/**
 * Start fetching toppar.
 *
 * Locality: toppar handler thread
 * Locks: none
 */
static void rd_kafka_toppar_fetch_start (rd_kafka_toppar_t *rktp,
					 int64_t offset,
					 rd_kafka_op_t *rko_orig) {
        rd_kafka_cgrp_t *rkcg = rko_orig->rko_u.fetch_start.rkcg;
        rd_kafka_resp_err_t err = 0;
        int32_t version = rko_orig->rko_version;

	rd_kafka_toppar_lock(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "FETCH",
                     "Start fetch for %.*s [%"PRId32"] in "
                     "state %s at offset %s (v%"PRId32")",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition,
                     rd_kafka_fetch_states[rktp->rktp_fetch_state],
                     rd_kafka_offset2str(offset), version);

        if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_STOPPING) {
                err = RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS;
		rd_kafka_toppar_unlock(rktp);
                goto err_reply;
        }

	rktp->rktp_op_version = version;

        if (rkcg) {
                rd_kafka_assert(rktp->rktp_rkt->rkt_rk, !rktp->rktp_cgrp);
                /* Attach toppar to cgrp */
                rktp->rktp_cgrp = rkcg;
                rd_kafka_cgrp_op(rkcg, rktp, RD_KAFKA_NO_REPLYQ,
                                 RD_KAFKA_OP_PARTITION_JOIN, 0);
        }


        if (offset == RD_KAFKA_OFFSET_BEGINNING ||
	    offset == RD_KAFKA_OFFSET_END ||
            offset <= RD_KAFKA_OFFSET_TAIL_BASE) {
		rd_kafka_toppar_next_offset_handle(rktp, offset);

	} else if (offset == RD_KAFKA_OFFSET_STORED) {
                rd_kafka_offset_store_init(rktp);

	} else if (offset == RD_KAFKA_OFFSET_INVALID) {
		rd_kafka_offset_reset(rktp, offset,
				      RD_KAFKA_RESP_ERR__NO_OFFSET,
				      "no previously committed offset "
				      "available");

	} else {
		rktp->rktp_next_offset = offset;
                rd_kafka_toppar_set_fetch_state(rktp,
						RD_KAFKA_TOPPAR_FETCH_ACTIVE);

                /* Wake-up broker thread which might be idling on IO */
                if (rktp->rktp_leader)
                        rd_kafka_broker_wakeup(rktp->rktp_leader);

	}

        rktp->rktp_offsets_fin.eof_offset = RD_KAFKA_OFFSET_INVALID;

	rd_kafka_toppar_unlock(rktp);

        /* Signal back to caller thread that start has commenced, or err */
err_reply:
        if (rko_orig->rko_replyq.q) {
                rd_kafka_op_t *rko;

                rko = rd_kafka_op_new(RD_KAFKA_OP_FETCH_START);

                rko->rko_err = err;
                rko->rko_rktp = rd_kafka_toppar_keep(rktp);

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




/**
 * Mark toppar's fetch state as stopped (all decommissioning is done,
 * offsets are stored, etc).
 *
 * Locality: toppar handler thread
 * Locks: toppar_lock(rktp) MUST be held
 */
void rd_kafka_toppar_fetch_stopped (rd_kafka_toppar_t *rktp,
                                    rd_kafka_resp_err_t err) {


        rd_kafka_toppar_set_fetch_state(rktp, RD_KAFKA_TOPPAR_FETCH_STOPPED);

        if (rktp->rktp_cgrp) {
                /* Detach toppar from cgrp */
                rd_kafka_cgrp_op(rktp->rktp_cgrp, rktp, RD_KAFKA_NO_REPLYQ,
                                 RD_KAFKA_OP_PARTITION_LEAVE, 0);
                rktp->rktp_cgrp = NULL;
        }

        /* Signal back to application thread that stop is done. */
	if (rktp->rktp_replyq.q) {
		rd_kafka_op_t *rko;
		rko = rd_kafka_op_new(RD_KAFKA_OP_FETCH_STOP|RD_KAFKA_OP_REPLY);
                rko->rko_err = err;
		rko->rko_rktp = rd_kafka_toppar_keep(rktp);

		rd_kafka_replyq_enq(&rktp->rktp_replyq, rko, 0);
	}
}


/**
 * Stop toppar fetcher.
 * This is usually an async operation.
 *
 * Locality: toppar handler thread
 */
void rd_kafka_toppar_fetch_stop (rd_kafka_toppar_t *rktp,
				 rd_kafka_op_t *rko_orig) {
        int32_t version = rko_orig->rko_version;

	rd_kafka_toppar_lock(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "FETCH",
                     "Stopping fetch for %.*s [%"PRId32"] in state %s (v%d)",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition,
                     rd_kafka_fetch_states[rktp->rktp_fetch_state], version);

	rktp->rktp_op_version = version;

	/* Abort pending offset lookups. */
	if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY)
		rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
				    &rktp->rktp_offset_query_tmr,
				    1/*lock*/);

        /* Clear out the forwarding queue. */
        rd_kafka_q_fwd_set(rktp->rktp_fetchq, NULL);

        /* Assign the future replyq to propagate stop results. */
        rd_kafka_assert(rktp->rktp_rkt->rkt_rk, rktp->rktp_replyq.q == NULL);
	if (rko_orig) {
		rktp->rktp_replyq = rko_orig->rko_replyq;
		rd_kafka_replyq_clear(&rko_orig->rko_replyq);
	}
        rd_kafka_toppar_set_fetch_state(rktp, RD_KAFKA_TOPPAR_FETCH_STOPPING);

        /* Stop offset store (possibly async).
         * NOTE: will call .._stopped() if store finishes immediately,
         *       so no more operations after this call! */
        rd_kafka_offset_store_stop(rktp);

	rd_kafka_toppar_unlock(rktp);
}


/**
 * Update a toppars offset.
 * The toppar must have been previously FETCH_START:ed
 *
 * Locality: toppar handler thread
 */
void rd_kafka_toppar_seek (rd_kafka_toppar_t *rktp,
			   int64_t offset, rd_kafka_op_t *rko_orig) {
        rd_kafka_resp_err_t err = 0;
        int32_t version = rko_orig->rko_version;

	rd_kafka_toppar_lock(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "FETCH",
                     "Seek %.*s [%"PRId32"] to offset %s "
                     "in state %s (v%"PRId32")",
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition,
		     rd_kafka_offset2str(offset),
                     rd_kafka_fetch_states[rktp->rktp_fetch_state], version);


        if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_STOPPING) {
                err = RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS;
                goto err_reply;
        } else if (!RD_KAFKA_TOPPAR_FETCH_IS_STARTED(rktp->rktp_fetch_state)) {
                err = RD_KAFKA_RESP_ERR__STATE;
                goto err_reply;
        } else if (offset == RD_KAFKA_OFFSET_STORED) {
		err = RD_KAFKA_RESP_ERR__INVALID_ARG;
		goto err_reply;
	}

	rktp->rktp_op_version = version;

	/* Abort pending offset lookups. */
	if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY)
		rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
				    &rktp->rktp_offset_query_tmr,
				    1/*lock*/);

	if (RD_KAFKA_OFFSET_IS_LOGICAL(offset))
		rd_kafka_toppar_next_offset_handle(rktp, offset);
	else {
		rktp->rktp_next_offset = offset;
                rd_kafka_toppar_set_fetch_state(rktp,
						RD_KAFKA_TOPPAR_FETCH_ACTIVE);

                /* Wake-up broker thread which might be idling on IO */
                if (rktp->rktp_leader)
                        rd_kafka_broker_wakeup(rktp->rktp_leader);
	}

        /* Signal back to caller thread that seek has commenced, or err */
err_reply:
	rd_kafka_toppar_unlock(rktp);

        if (rko_orig && rko_orig->rko_replyq.q) {
                rd_kafka_op_t *rko;

                rko = rd_kafka_op_new(RD_KAFKA_OP_SEEK|RD_KAFKA_OP_REPLY);

                rko->rko_err = err;
		rko->rko_u.fetch_start.offset =
			rko_orig->rko_u.fetch_start.offset;
                rko->rko_rktp = rd_kafka_toppar_keep(rktp);

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


static void rd_kafka_toppar_pause_resume (rd_kafka_toppar_t *rktp,
					  rd_kafka_op_t *rko_orig) {
	rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
	int pause = rko_orig->rko_u.pause.pause;
	int flag = rko_orig->rko_u.pause.flag;
        int32_t version = rko_orig->rko_version;

	rd_kafka_toppar_lock(rktp);

	rktp->rktp_op_version = version;

	if (pause) {
		/* Pause partition */
		rktp->rktp_flags |= flag;

		if (rk->rk_type == RD_KAFKA_CONSUMER) {
			/* Save offset of last consumed message+1 as the
			 * next message to fetch on resume. */
			if (rktp->rktp_app_offset != RD_KAFKA_OFFSET_INVALID) {
				rktp->rktp_next_offset = rktp->rktp_app_offset;
			}

			rd_kafka_dbg(rk, TOPIC, pause?"PAUSE":"RESUME",
				     "%s %s [%"PRId32"]: at offset %s "
				     "(state %s, v%d)",
				     pause ? "Pause":"Resume",
				     rktp->rktp_rkt->rkt_topic->str,
				     rktp->rktp_partition,
				     rd_kafka_offset2str(
					     rktp->rktp_next_offset),
				     rd_kafka_fetch_states[rktp->
							   rktp_fetch_state],
				     version);
		} else {
			rd_kafka_dbg(rk, TOPIC, pause?"PAUSE":"RESUME",
				     "%s %s [%"PRId32"] (state %s, v%d)",
				     pause ? "Pause":"Resume",
				     rktp->rktp_rkt->rkt_topic->str,
				     rktp->rktp_partition,
				     rd_kafka_fetch_states[rktp->
							   rktp_fetch_state],
				     version);
			}

	} else {
		/* Resume partition */
		rktp->rktp_flags &= ~flag;

		if (rk->rk_type == RD_KAFKA_CONSUMER) {
			rd_kafka_dbg(rk, TOPIC, pause?"PAUSE":"RESUME",
				     "%s %s [%"PRId32"]: at offset %s "
				     "(state %s, v%d)",
				     rktp->rktp_fetch_state ==
				     RD_KAFKA_TOPPAR_FETCH_ACTIVE ?
				     "Resuming" : "Not resuming stopped",
				     rktp->rktp_rkt->rkt_topic->str,
				     rktp->rktp_partition,
				     rd_kafka_offset2str(
					     rktp->rktp_next_offset),
				     rd_kafka_fetch_states[rktp->
							   rktp_fetch_state],
				     version);

			/* If the resuming offset is logical we
			 * need to trigger a seek (that performs the
			 * logical->absolute lookup logic) to get
			 * things going.
			 * Typical case is when a partition is paused
			 * before anything has been consumed by app
			 * yet thus having rktp_app_offset=INVALID. */
			if ((rktp->rktp_fetch_state ==
			     RD_KAFKA_TOPPAR_FETCH_ACTIVE ||
			     rktp->rktp_fetch_state ==
			     RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT) &&
			    rktp->rktp_next_offset == RD_KAFKA_OFFSET_INVALID)
				rd_kafka_toppar_next_offset_handle(
					rktp, rktp->rktp_next_offset);

		} else
			rd_kafka_dbg(rk, TOPIC, pause?"PAUSE":"RESUME",
				     "%s %s [%"PRId32"] (state %s, v%d)",
				     pause ? "Pause":"Resume",
				     rktp->rktp_rkt->rkt_topic->str,
				     rktp->rktp_partition,
				     rd_kafka_fetch_states[rktp->
							   rktp_fetch_state],
				     version);
	}
	rd_kafka_toppar_unlock(rktp);

	if (pause && rk->rk_type == RD_KAFKA_CONSUMER) {
		/* Flush partition's fetch queue */
		rd_kafka_q_purge_toppar_version(rktp->rktp_fetchq, rktp,
						rko_orig->rko_version);
	}
}




/**
 * @brief Decide whether this toppar should be on the fetch list or not.
 *
 * Also:
 *  - update toppar's op version (for broker thread's copy)
 *  - finalize statistics (move rktp_offsets to rktp_offsets_fin)
 *
 * @returns the partition's Fetch backoff timestamp, or 0 if no backoff.
 *
 * @locality broker thread
 */
rd_ts_t rd_kafka_toppar_fetch_decide (rd_kafka_toppar_t *rktp,
				   rd_kafka_broker_t *rkb,
				   int force_remove) {
        int should_fetch = 1;
        const char *reason = "";
        int32_t version;
        rd_ts_t ts_backoff = 0;

	rd_kafka_toppar_lock(rktp);

	/* Forced removal from fetch list */
	if (unlikely(force_remove)) {
		reason = "forced removal";
		should_fetch = 0;
		goto done;
	}

	if (unlikely((rktp->rktp_flags & RD_KAFKA_TOPPAR_F_REMOVE) != 0)) {
		reason = "partition removed";
		should_fetch = 0;
		goto done;
	}

	/* Skip toppars not in active fetch state */
	if (rktp->rktp_fetch_state != RD_KAFKA_TOPPAR_FETCH_ACTIVE) {
                reason = "not in active fetch state";
		should_fetch = 0;
		goto done;
	}

        /* Update broker thread's fetch op version */
        version = rktp->rktp_op_version;
        if (version > rktp->rktp_fetch_version ||
	    rktp->rktp_next_offset != rktp->rktp_last_next_offset) {
                /* New version barrier, something was modified from the
                 * control plane. Reset and start over.
		 * Alternatively only the next_offset changed but not the
		 * barrier, which is the case when automatically triggering
		 * offset.reset (such as on PARTITION_EOF). */

                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "FETCHDEC",
                             "Topic %s [%"PRId32"]: fetch decide: "
                             "updating to version %d (was %d) at "
                             "offset %"PRId64" (was %"PRId64")",
                             rktp->rktp_rkt->rkt_topic->str,
                             rktp->rktp_partition,
                             version, rktp->rktp_fetch_version,
                             rktp->rktp_next_offset,
                             rktp->rktp_offsets.fetch_offset);

                rd_kafka_offset_stats_reset(&rktp->rktp_offsets);

                /* New start offset */
                rktp->rktp_offsets.fetch_offset = rktp->rktp_next_offset;
		rktp->rktp_last_next_offset = rktp->rktp_next_offset;

                rktp->rktp_fetch_version = version;

                rd_kafka_q_purge_toppar_version(rktp->rktp_fetchq, rktp,
                                                version);
        }


	if (RD_KAFKA_TOPPAR_IS_PAUSED(rktp)) {
		should_fetch = 0;
		reason = "paused";

	} else if (RD_KAFKA_OFFSET_IS_LOGICAL(rktp->rktp_next_offset)) {
                should_fetch = 0;
                reason = "no concrete offset";

        } else if (rd_kafka_q_len(rktp->rktp_fetchq) >=
		   rkb->rkb_rk->rk_conf.queued_min_msgs) {
		/* Skip toppars who's local message queue is already above
		 * the lower threshold. */
                reason = "queued.min.messages exceeded";
                should_fetch = 0;

        } else if ((int64_t)rd_kafka_q_size(rktp->rktp_fetchq) >=
            rkb->rkb_rk->rk_conf.queued_max_msg_bytes) {
                reason = "queued.max.messages.kbytes exceeded";
                should_fetch = 0;

        } else if (rktp->rktp_ts_fetch_backoff > rd_clock()) {
                reason = "fetch backed off";
                ts_backoff = rktp->rktp_ts_fetch_backoff;
                should_fetch = 0;
        }

 done:
        /* Copy offset stats to finalized place holder. */
        rktp->rktp_offsets_fin = rktp->rktp_offsets;

        if (rktp->rktp_fetch != should_fetch) {
                rd_rkb_dbg(rkb, FETCH, "FETCH",
                           "Topic %s [%"PRId32"] in state %s at offset %s "
                           "(%d/%d msgs, %"PRId64"/%d kb queued, "
			   "opv %"PRId32") is %sfetchable: %s",
                           rktp->rktp_rkt->rkt_topic->str,
                           rktp->rktp_partition,
			   rd_kafka_fetch_states[rktp->rktp_fetch_state],
                           rd_kafka_offset2str(rktp->rktp_next_offset),
                           rd_kafka_q_len(rktp->rktp_fetchq),
                           rkb->rkb_rk->rk_conf.queued_min_msgs,
                           rd_kafka_q_size(rktp->rktp_fetchq) / 1024,
                           rkb->rkb_rk->rk_conf.queued_max_msg_kbytes,
			   rktp->rktp_fetch_version,
                           should_fetch ? "" : "not ", reason);

                if (should_fetch) {
			rd_dassert(rktp->rktp_fetch_version > 0);
                        rd_kafka_broker_active_toppar_add(rkb, rktp);
                } else {
                        rd_kafka_broker_active_toppar_del(rkb, rktp);
                        /* Non-fetching partitions will have an
                         * indefinate backoff, unless explicitly specified. */
                        if (!ts_backoff)
                                ts_backoff = RD_TS_MAX;
                }
        }

        rd_kafka_toppar_unlock(rktp);

        return ts_backoff;
}


/**
 * @brief Serve a toppar in a consumer broker thread.
 *        This is considered the fast path and should be minimal,
 *        mostly focusing on fetch related mechanisms.
 *
 * @returns the partition's Fetch backoff timestamp, or 0 if no backoff.
 *
 * @locality broker thread
 * @locks none
 */
rd_ts_t rd_kafka_broker_consumer_toppar_serve (rd_kafka_broker_t *rkb,
                                               rd_kafka_toppar_t *rktp) {
        return rd_kafka_toppar_fetch_decide(rktp, rkb, 0);
}



/**
 * Serve a toppar op
 * 'rktp' may be NULL for certain ops (OP_RECV_BUF)
 *
 * @locality toppar handler thread
 */
static rd_kafka_op_res_t
rd_kafka_toppar_op_serve (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_toppar_t *rktp = NULL;
	int outdated = 0;

	if (rko->rko_rktp)
		rktp = rd_kafka_toppar_s2i(rko->rko_rktp);

	if (rktp) {
		outdated = rd_kafka_op_version_outdated(rko,
							rktp->rktp_op_version);

		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "OP",
			     "%.*s [%"PRId32"] received %sop %s "
			     "(v%"PRId32") in fetch-state %s (opv%d)",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition,
			     outdated ? "outdated ": "",
			     rd_kafka_op2str(rko->rko_type),
			     rko->rko_version,
			     rd_kafka_fetch_states[rktp->rktp_fetch_state],
			     rktp->rktp_op_version);

		if (outdated) {
#if ENABLE_DEVEL
			rd_kafka_op_print(stdout, "PART_OUTDATED", rko);
#endif
                        rd_kafka_op_destroy(rko);
			return RD_KAFKA_OP_RES_HANDLED;
		}
	}

	switch ((int)rko->rko_type)
	{
	case RD_KAFKA_OP_FETCH_START:
		rd_kafka_toppar_fetch_start(rktp,
					    rko->rko_u.fetch_start.offset, rko);
		break;

	case RD_KAFKA_OP_FETCH_STOP:
		rd_kafka_toppar_fetch_stop(rktp, rko);
		break;

	case RD_KAFKA_OP_SEEK:
		rd_kafka_toppar_seek(rktp, rko->rko_u.fetch_start.offset, rko);
		break;

	case RD_KAFKA_OP_PAUSE:
		rd_kafka_toppar_pause_resume(rktp, rko);
		break;

        case RD_KAFKA_OP_OFFSET_COMMIT | RD_KAFKA_OP_REPLY:
                rd_kafka_assert(NULL, rko->rko_u.offset_commit.cb);
                rko->rko_u.offset_commit.cb(
                        rk, rko->rko_err,
                        rko->rko_u.offset_commit.partitions,
                        rko->rko_u.offset_commit.opaque);
                break;

	case RD_KAFKA_OP_OFFSET_FETCH | RD_KAFKA_OP_REPLY:
        {
                /* OffsetFetch reply */
                rd_kafka_topic_partition_list_t *offsets =
			rko->rko_u.offset_fetch.partitions;
                shptr_rd_kafka_toppar_t *s_rktp;
		int64_t offset = RD_KAFKA_OFFSET_INVALID;

                s_rktp = offsets->elems[0]._private;
                if (!rko->rko_err) {
                        /* Request succeeded but per-partition might have failed */
                        rko->rko_err = offsets->elems[0].err;
			offset       = offsets->elems[0].offset;
                }
                offsets->elems[0]._private = NULL;
                rd_kafka_topic_partition_list_destroy(offsets);
		rko->rko_u.offset_fetch.partitions = NULL;
                rktp = rd_kafka_toppar_s2i(s_rktp);

		rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
				    &rktp->rktp_offset_query_tmr,
				    1/*lock*/);

		rd_kafka_toppar_lock(rktp);

		if (rko->rko_err) {
			rd_kafka_dbg(rktp->rktp_rkt->rkt_rk,
				     TOPIC, "OFFSET",
				     "Failed to fetch offset for "
				     "%.*s [%"PRId32"]: %s",
				     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
				     rktp->rktp_partition,
				     rd_kafka_err2str(rko->rko_err));

			/* Keep on querying until we succeed. */
			rd_kafka_toppar_set_fetch_state(rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);

			rd_kafka_toppar_unlock(rktp);

			rd_kafka_timer_start(&rktp->rktp_rkt->rkt_rk->rk_timers,
					     &rktp->rktp_offset_query_tmr,
					     500*1000,
					     rd_kafka_offset_query_tmr_cb,
					     rktp);

			/* Propagate error to application */
			if (rko->rko_err != RD_KAFKA_RESP_ERR__WAIT_COORD) {
				rd_kafka_q_op_err(rktp->rktp_fetchq,
						  RD_KAFKA_OP_ERR, rko->rko_err,
						  0, rktp, 0,
						  "Failed to fetch "
						  "offsets from brokers: %s",
						  rd_kafka_err2str(rko->rko_err));
			}

			rd_kafka_toppar_destroy(s_rktp);

			break;
		}

		rd_kafka_dbg(rktp->rktp_rkt->rkt_rk,
			     TOPIC, "OFFSET",
			     "%.*s [%"PRId32"]: OffsetFetch returned "
			     "offset %s (%"PRId64")",
			     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
			     rktp->rktp_partition,
			     rd_kafka_offset2str(offset), offset);

		if (offset > 0)
			rktp->rktp_committed_offset = offset;

		if (offset >= 0)
			rd_kafka_toppar_next_offset_handle(rktp, offset);
		else
			rd_kafka_offset_reset(rktp, offset,
					      RD_KAFKA_RESP_ERR__NO_OFFSET,
					      "no previously committed offset "
					      "available");
		rd_kafka_toppar_unlock(rktp);

                rd_kafka_toppar_destroy(s_rktp);
        }
        break;

        default:
                rd_kafka_assert(NULL, !*"unknown type");
                break;
        }

        rd_kafka_op_destroy(rko);

        return RD_KAFKA_OP_RES_HANDLED;
}





/**
 * Send command op to toppar (handled by toppar's thread).
 *
 * Locality: any thread
 */
static void rd_kafka_toppar_op0 (rd_kafka_toppar_t *rktp, rd_kafka_op_t *rko,
				 rd_kafka_replyq_t replyq) {
        rko->rko_rktp = rd_kafka_toppar_keep(rktp);
	rko->rko_replyq = replyq;

        rd_kafka_q_enq(rktp->rktp_ops, rko);
}


/**
 * Send command op to toppar (handled by toppar's thread).
 *
 * Locality: any thread
 */
static void rd_kafka_toppar_op (rd_kafka_toppar_t *rktp,
				rd_kafka_op_type_t type, int32_t version,
				int64_t offset, rd_kafka_cgrp_t *rkcg,
				rd_kafka_replyq_t replyq) {
        rd_kafka_op_t *rko;

        rko = rd_kafka_op_new(type);
	rko->rko_version = version;
        if (type == RD_KAFKA_OP_FETCH_START ||
	    type == RD_KAFKA_OP_SEEK) {
		if (rkcg)
			rko->rko_u.fetch_start.rkcg = rkcg;
		rko->rko_u.fetch_start.offset = offset;
	}

	rd_kafka_toppar_op0(rktp, rko, replyq);
}



/**
 * Start consuming partition (async operation).
 *  'offset' is the initial offset
 *  'fwdq' is an optional queue to forward messages to, if this is NULL
 *  then messages will be enqueued on rktp_fetchq.
 *  'replyq' is an optional queue for handling the consume_start ack.
 *
 * This is the thread-safe interface that can be called from any thread.
 */
rd_kafka_resp_err_t rd_kafka_toppar_op_fetch_start (rd_kafka_toppar_t *rktp,
                                                    int64_t offset,
                                                    rd_kafka_q_t *fwdq,
                                                    rd_kafka_replyq_t replyq) {
	int32_t version;

        rd_kafka_q_lock(rktp->rktp_fetchq);
        if (fwdq && !(rktp->rktp_fetchq->rkq_flags & RD_KAFKA_Q_F_FWD_APP))
                rd_kafka_q_fwd_set0(rktp->rktp_fetchq, fwdq,
                                    0, /* no do_lock */
                                    0 /* no fwd_app */);
        rd_kafka_q_unlock(rktp->rktp_fetchq);

	/* Bump version barrier. */
	version = rd_kafka_toppar_version_new_barrier(rktp);

	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "CONSUMER",
		     "Start consuming %.*s [%"PRId32"] at "
		     "offset %s (v%"PRId32")",
		     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
		     rktp->rktp_partition, rd_kafka_offset2str(offset),
		     version);

        rd_kafka_toppar_op(rktp, RD_KAFKA_OP_FETCH_START, version,
                           offset, rktp->rktp_rkt->rkt_rk->rk_cgrp, replyq);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * Stop consuming partition (async operatoin)
 * This is thread-safe interface that can be called from any thread.
 *
 * Locality: any thread
 */
rd_kafka_resp_err_t rd_kafka_toppar_op_fetch_stop (rd_kafka_toppar_t *rktp,
                                                   rd_kafka_replyq_t replyq) {
	int32_t version;

	/* Bump version barrier. */
        version = rd_kafka_toppar_version_new_barrier(rktp);

        rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "CONSUMER",
		     "Stop consuming %.*s [%"PRId32"] (v%"PRId32")",
		     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
		     rktp->rktp_partition, version);

        rd_kafka_toppar_op(rktp, RD_KAFKA_OP_FETCH_STOP, version,
			   0, NULL, replyq);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * Set/Seek offset of a consumed partition (async operation).
 *  'offset' is the target offset
 *  'replyq' is an optional queue for handling the ack.
 *
 * This is the thread-safe interface that can be called from any thread.
 */
rd_kafka_resp_err_t rd_kafka_toppar_op_seek (rd_kafka_toppar_t *rktp,
                                             int64_t offset,
                                             rd_kafka_replyq_t replyq) {
	int32_t version;

	/* Bump version barrier. */
	version = rd_kafka_toppar_version_new_barrier(rktp);

	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "CONSUMER",
		     "Seek %.*s [%"PRId32"] to "
		     "offset %s (v%"PRId32")",
		     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
		     rktp->rktp_partition, rd_kafka_offset2str(offset),
		     version);

        rd_kafka_toppar_op(rktp, RD_KAFKA_OP_SEEK, version,
			   offset, NULL, replyq);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * Pause/resume partition (async operation).
 * \p flag is either RD_KAFKA_TOPPAR_F_APP_PAUSE or .._F_LIB_PAUSE
 * depending on if the app paused or librdkafka.
 * \p pause is 1 for pausing or 0 for resuming.
 *
 * Locality: any
 */
static rd_kafka_resp_err_t
rd_kafka_toppar_op_pause_resume (rd_kafka_toppar_t *rktp,
				 int pause, int flag) {
	int32_t version;
	rd_kafka_op_t *rko;

	/* Bump version barrier. */
	version = rd_kafka_toppar_version_new_barrier(rktp);

	rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, pause ? "PAUSE":"RESUME",
		     "%s %.*s [%"PRId32"] (v%"PRId32")",
		     pause ? "Pause" : "Resume",
		     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
		     rktp->rktp_partition, version);

	rko = rd_kafka_op_new(RD_KAFKA_OP_PAUSE);
	rko->rko_version = version;
	rko->rko_u.pause.pause = pause;
	rko->rko_u.pause.flag = flag;

	rd_kafka_toppar_op0(rktp, rko, RD_KAFKA_NO_REPLYQ);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}





/**
 * Pause or resume a list of partitions.
 * \p flag is either RD_KAFKA_TOPPAR_F_APP_PAUSE or .._F_LIB_PAUSE
 * depending on if the app paused or librdkafka.
 * \p pause is 1 for pausing or 0 for resuming.
 *
 * Locality: any
 *
 * @remark This is an asynchronous call, the actual pause/resume is performed
 *         by toppar_pause() in the toppar's handler thread.
 */
rd_kafka_resp_err_t
rd_kafka_toppars_pause_resume (rd_kafka_t *rk, int pause, int flag,
			       rd_kafka_topic_partition_list_t *partitions) {
	int i;

	rd_kafka_dbg(rk, TOPIC, pause ? "PAUSE":"RESUME",
		     "%s %s %d partition(s)",
		     flag & RD_KAFKA_TOPPAR_F_APP_PAUSE ? "Application" : "Library",
		     pause ? "pausing" : "resuming", partitions->cnt);

	for (i = 0 ; i < partitions->cnt ; i++) {
		rd_kafka_topic_partition_t *rktpar = &partitions->elems[i];
		shptr_rd_kafka_toppar_t *s_rktp;
		rd_kafka_toppar_t *rktp;

                s_rktp = rd_kafka_topic_partition_list_get_toppar(rk, rktpar);
		if (!s_rktp) {
			rd_kafka_dbg(rk, TOPIC, pause ? "PAUSE":"RESUME",
				     "%s %s [%"PRId32"]: skipped: "
				     "unknown partition",
				     pause ? "Pause":"Resume",
				     rktpar->topic, rktpar->partition);

			rktpar->err = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;
			continue;
		}

		rktp = rd_kafka_toppar_s2i(s_rktp);

		rd_kafka_toppar_op_pause_resume(rktp, pause, flag);

		rd_kafka_toppar_destroy(s_rktp);

		rktpar->err = RD_KAFKA_RESP_ERR_NO_ERROR;
	}

	return RD_KAFKA_RESP_ERR_NO_ERROR;
}





/**
 * Propagate error for toppar
 */
void rd_kafka_toppar_enq_error (rd_kafka_toppar_t *rktp,
                                rd_kafka_resp_err_t err,
                                const char *reason) {
        rd_kafka_op_t *rko;
        char buf[512];

        rko = rd_kafka_op_new(RD_KAFKA_OP_ERR);
        rko->rko_err  = err;
        rko->rko_rktp = rd_kafka_toppar_keep(rktp);

        rd_snprintf(buf, sizeof(buf), "%.*s [%"PRId32"]: %s (%s)",
                    RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                    rktp->rktp_partition, reason,
                    rd_kafka_err2str(err));

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

        rd_kafka_q_enq(rktp->rktp_fetchq, rko);
}





/**
 * Returns the local leader broker for this toppar.
 * If \p proper_broker is set NULL will be returned if current handler
 * is not a proper broker (INTERNAL broker).
 *
 * The returned broker has an increased refcount.
 *
 * Locks: none
 */
rd_kafka_broker_t *rd_kafka_toppar_leader (rd_kafka_toppar_t *rktp,
                                           int proper_broker) {
        rd_kafka_broker_t *rkb;
        rd_kafka_toppar_lock(rktp);
        rkb = rktp->rktp_leader;
        if (rkb) {
                if (proper_broker && rkb->rkb_source == RD_KAFKA_INTERNAL)
                        rkb = NULL;
                else
                        rd_kafka_broker_keep(rkb);
        }
        rd_kafka_toppar_unlock(rktp);

        return rkb;
}


/**
 * @brief Take action when partition leader becomes unavailable.
 *        This should be called when leader-specific requests fail with
 *        NOT_LEADER_FOR.. or similar error codes, e.g. ProduceRequest.
 *
 * @locks none
 * @locality any
 */
void rd_kafka_toppar_leader_unavailable (rd_kafka_toppar_t *rktp,
                                         const char *reason,
                                         rd_kafka_resp_err_t err) {
        rd_kafka_itopic_t *rkt = rktp->rktp_rkt;

        rd_kafka_dbg(rkt->rkt_rk, TOPIC, "LEADERUA",
                     "%s [%"PRId32"]: leader unavailable: %s: %s",
                     rkt->rkt_topic->str, rktp->rktp_partition, reason,
                     rd_kafka_err2str(err));

        rd_kafka_topic_wrlock(rkt);
        rkt->rkt_flags |= RD_KAFKA_TOPIC_F_LEADER_UNAVAIL;
        rd_kafka_topic_wrunlock(rkt);

        rd_kafka_topic_fast_leader_query(rkt->rkt_rk);
}


const char *
rd_kafka_topic_partition_topic (const rd_kafka_topic_partition_t *rktpar) {
        const rd_kafka_toppar_t *rktp = (const rd_kafka_toppar_t *)rktpar;
        return rktp->rktp_rkt->rkt_topic->str;
}

int32_t
rd_kafka_topic_partition_partition (const rd_kafka_topic_partition_t *rktpar) {
        const rd_kafka_toppar_t *rktp = (const rd_kafka_toppar_t *)rktpar;
        return rktp->rktp_partition;
}

void rd_kafka_topic_partition_get (const rd_kafka_topic_partition_t *rktpar,
                                   const char **name, int32_t *partition) {
        const rd_kafka_toppar_t *rktp = (const rd_kafka_toppar_t *)rktpar;
        *name = rktp->rktp_rkt->rkt_topic->str;
        *partition = rktp->rktp_partition;
}




/**
 *
 * rd_kafka_topic_partition_t lists
 * Fixed-size non-growable list of partitions for propagation to application.
 *
 */


static void
rd_kafka_topic_partition_list_grow (rd_kafka_topic_partition_list_t *rktparlist,
                                    int add_size) {
        if (add_size < rktparlist->size)
                add_size = RD_MAX(rktparlist->size, 32);

        rktparlist->size += add_size;
        rktparlist->elems = rd_realloc(rktparlist->elems,
                                       sizeof(*rktparlist->elems) *
                                       rktparlist->size);

}
/**
 * Create a list for fitting 'size' topic_partitions (rktp).
 */
rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_new (int size) {
        rd_kafka_topic_partition_list_t *rktparlist;

        rktparlist = rd_calloc(1, sizeof(*rktparlist));

        rktparlist->size = size;
        rktparlist->cnt = 0;

        if (size > 0)
                rd_kafka_topic_partition_list_grow(rktparlist, size);

        return rktparlist;
}



rd_kafka_topic_partition_t *rd_kafka_topic_partition_new (const char *topic,
							  int32_t partition) {
	rd_kafka_topic_partition_t *rktpar = rd_calloc(1, sizeof(*rktpar));

	rktpar->topic = rd_strdup(topic);
	rktpar->partition = partition;

	return rktpar;
}


rd_kafka_topic_partition_t *
rd_kafka_topic_partition_new_from_rktp (rd_kafka_toppar_t *rktp) {
	rd_kafka_topic_partition_t *rktpar = rd_calloc(1, sizeof(*rktpar));

	rktpar->topic = RD_KAFKAP_STR_DUP(rktp->rktp_rkt->rkt_topic);
	rktpar->partition = rktp->rktp_partition;

	return rktpar;
}



static void
rd_kafka_topic_partition_destroy0 (rd_kafka_topic_partition_t *rktpar, int do_free) {
	if (rktpar->topic)
		rd_free(rktpar->topic);
	if (rktpar->metadata)
		rd_free(rktpar->metadata);
	if (rktpar->_private)
		rd_kafka_toppar_destroy((shptr_rd_kafka_toppar_t *)
					rktpar->_private);

	if (do_free)
		rd_free(rktpar);
}

void rd_kafka_topic_partition_destroy (rd_kafka_topic_partition_t *rktpar) {
	rd_kafka_topic_partition_destroy0(rktpar, 1);
}


/**
 * Destroys a list previously created with .._list_new() and drops
 * any references to contained toppars.
 */
void
rd_kafka_topic_partition_list_destroy (rd_kafka_topic_partition_list_t *rktparlist) {
        int i;

        for (i = 0 ; i < rktparlist->cnt ; i++)
		rd_kafka_topic_partition_destroy0(&rktparlist->elems[i], 0);

        if (rktparlist->elems)
                rd_free(rktparlist->elems);

        rd_free(rktparlist);
}


/**
 * Add a partition to an rktpar list.
 * The list must have enough room to fit it.
 *
 * '_private' must be NULL or a valid 'shptr_rd_kafka_toppar_t *'.
 *
 * Returns a pointer to the added element.
 */
rd_kafka_topic_partition_t *
rd_kafka_topic_partition_list_add0 (rd_kafka_topic_partition_list_t *rktparlist,
                                    const char *topic, int32_t partition,
				    shptr_rd_kafka_toppar_t *_private) {
        rd_kafka_topic_partition_t *rktpar;
        if (rktparlist->cnt == rktparlist->size)
                rd_kafka_topic_partition_list_grow(rktparlist, 1);
        rd_kafka_assert(NULL, rktparlist->cnt < rktparlist->size);

        rktpar = &rktparlist->elems[rktparlist->cnt++];
        memset(rktpar, 0, sizeof(*rktpar));
        rktpar->topic = rd_strdup(topic);
        rktpar->partition = partition;
	rktpar->offset = RD_KAFKA_OFFSET_INVALID;
        rktpar->_private = _private;

        return rktpar;
}


rd_kafka_topic_partition_t *
rd_kafka_topic_partition_list_add (rd_kafka_topic_partition_list_t *rktparlist,
                                   const char *topic, int32_t partition) {
        return rd_kafka_topic_partition_list_add0(rktparlist,
                                                  topic, partition, NULL);
}


/**
 * Adds a consecutive list of partitions to a list
 */
void
rd_kafka_topic_partition_list_add_range (rd_kafka_topic_partition_list_t
                                         *rktparlist,
                                         const char *topic,
                                         int32_t start, int32_t stop) {

        for (; start <= stop ; start++)
                rd_kafka_topic_partition_list_add(rktparlist, topic, start);
}


rd_kafka_topic_partition_t *
rd_kafka_topic_partition_list_upsert (
        rd_kafka_topic_partition_list_t *rktparlist,
        const char *topic, int32_t partition) {
        rd_kafka_topic_partition_t *rktpar;

        if ((rktpar = rd_kafka_topic_partition_list_find(rktparlist,
                                                         topic, partition)))
                return rktpar;

        return rd_kafka_topic_partition_list_add(rktparlist, topic, partition);
}

/**
 * @brief Creates a copy of \p rktpar and adds it to \p rktparlist
 */
void
rd_kafka_topic_partition_copy (rd_kafka_topic_partition_list_t *rktparlist,
                               const rd_kafka_topic_partition_t *rktpar) {
        rd_kafka_topic_partition_t *dst;

        dst = rd_kafka_topic_partition_list_add0(
                rktparlist,
                rktpar->topic,
                rktpar->partition,
                rktpar->_private ?
                rd_kafka_toppar_keep(
                        rd_kafka_toppar_s2i((shptr_rd_kafka_toppar_t *)
                                            rktpar->_private)) : NULL);
        dst->offset = rktpar->offset;
        dst->opaque = rktpar->opaque;
        dst->err    = rktpar->err;
        if (rktpar->metadata_size > 0) {
                dst->metadata =
                        rd_malloc(rktpar->metadata_size);
                dst->metadata_size = rktpar->metadata_size;
                memcpy((void *)dst->metadata, rktpar->metadata,
                       rktpar->metadata_size);
        }
}



/**
 * Create and return a copy of list 'src'
 */
rd_kafka_topic_partition_list_t *
rd_kafka_topic_partition_list_copy (const rd_kafka_topic_partition_list_t *src){
        rd_kafka_topic_partition_list_t *dst;
        int i;

        dst = rd_kafka_topic_partition_list_new(src->size);

        for (i = 0 ; i < src->cnt ; i++)
                rd_kafka_topic_partition_copy(dst, &src->elems[i]);
        return dst;
}

/**
 * @returns (and sets if necessary) the \p rktpar's _private / toppar.
 * @remark a new reference is returned.
 */
shptr_rd_kafka_toppar_t *
rd_kafka_topic_partition_get_toppar (rd_kafka_t *rk,
                                     rd_kafka_topic_partition_t *rktpar) {
        shptr_rd_kafka_toppar_t *s_rktp;

        if (!(s_rktp = rktpar->_private))
                s_rktp = rktpar->_private =
                        rd_kafka_toppar_get2(rk,
                                             rktpar->topic,
                                             rktpar->partition, 0, 0);
        if (!s_rktp)
                return NULL;

        return rd_kafka_toppar_keep(rd_kafka_toppar_s2i(s_rktp));
}


static int rd_kafka_topic_partition_cmp (const void *_a, const void *_b,
                                         void *opaque) {
        const rd_kafka_topic_partition_t *a = _a;
        const rd_kafka_topic_partition_t *b = _b;
        int r = strcmp(a->topic, b->topic);
        if (r)
                return r;
        else
                return a->partition - b->partition;
}


/**
 * @brief Search 'rktparlist' for 'topic' and 'partition'.
 * @returns the elems[] index or -1 on miss.
 */
int
rd_kafka_topic_partition_list_find0 (rd_kafka_topic_partition_list_t *rktparlist,
				     const char *topic, int32_t partition) {
        rd_kafka_topic_partition_t skel;
        int i;

        skel.topic = (char *)topic;
        skel.partition = partition;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                if (!rd_kafka_topic_partition_cmp(&skel,
                                                  &rktparlist->elems[i],
                                                  NULL))
                        return i;
        }

        return -1;
}

rd_kafka_topic_partition_t *
rd_kafka_topic_partition_list_find (rd_kafka_topic_partition_list_t *rktparlist,
				     const char *topic, int32_t partition) {
	int i = rd_kafka_topic_partition_list_find0(rktparlist,
						    topic, partition);
	if (i == -1)
		return NULL;
	else
		return &rktparlist->elems[i];
}


int
rd_kafka_topic_partition_list_del_by_idx (rd_kafka_topic_partition_list_t *rktparlist,
					  int idx) {
	if (unlikely(idx < 0 || idx >= rktparlist->cnt))
		return 0;

	rktparlist->cnt--;
	rd_kafka_topic_partition_destroy0(&rktparlist->elems[idx], 0);
	memmove(&rktparlist->elems[idx], &rktparlist->elems[idx+1],
		(rktparlist->cnt - idx) * sizeof(rktparlist->elems[idx]));

	return 1;
}


int
rd_kafka_topic_partition_list_del (rd_kafka_topic_partition_list_t *rktparlist,
				   const char *topic, int32_t partition) {
	int i = rd_kafka_topic_partition_list_find0(rktparlist,
						    topic, partition);
	if (i == -1)
		return 0;

	return rd_kafka_topic_partition_list_del_by_idx(rktparlist, i);
}



/**
 * Returns true if 'topic' matches the 'rktpar', else false.
 * On match, if rktpar is a regex pattern then 'matched_by_regex' is set to 1.
 */
int rd_kafka_topic_partition_match (rd_kafka_t *rk,
				    const rd_kafka_group_member_t *rkgm,
				    const rd_kafka_topic_partition_t *rktpar,
				    const char *topic, int *matched_by_regex) {
	int ret = 0;

	if (*rktpar->topic == '^') {
		char errstr[128];

		ret = rd_regex_match(rktpar->topic, topic,
				     errstr, sizeof(errstr));
		if (ret == -1) {
			rd_kafka_dbg(rk, CGRP,
				     "SUBMATCH",
				     "Invalid regex for member "
				     "\"%.*s\" subscription \"%s\": %s",
				     RD_KAFKAP_STR_PR(rkgm->rkgm_member_id),
				     rktpar->topic, errstr);
			return 0;
		}

		if (ret && matched_by_regex)
			*matched_by_regex = 1;

	} else if (!strcmp(rktpar->topic, topic)) {

		if (matched_by_regex)
			*matched_by_regex = 0;

		ret = 1;
	}

	return ret;
}



void rd_kafka_topic_partition_list_sort (
        rd_kafka_topic_partition_list_t *rktparlist,
        int (*cmp) (const void *, const void *, void *),
        void *opaque) {

        if (!cmp)
                cmp = rd_kafka_topic_partition_cmp;

        rd_qsort_r(rktparlist->elems, rktparlist->cnt,
                   sizeof(*rktparlist->elems),
                   cmp, opaque);
}


void rd_kafka_topic_partition_list_sort_by_topic (
        rd_kafka_topic_partition_list_t *rktparlist) {
        rd_kafka_topic_partition_list_sort(rktparlist,
                                           rd_kafka_topic_partition_cmp, NULL);
}

rd_kafka_resp_err_t rd_kafka_topic_partition_list_set_offset (
	rd_kafka_topic_partition_list_t *rktparlist,
	const char *topic, int32_t partition, int64_t offset) {
	rd_kafka_topic_partition_t *rktpar;

	if (!(rktpar = rd_kafka_topic_partition_list_find(rktparlist,
							  topic, partition)))
		return RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;

	rktpar->offset = offset;

	return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * @brief Reset all offsets to the provided value.
 */
void
rd_kafka_topic_partition_list_reset_offsets (rd_kafka_topic_partition_list_t *rktparlist,
					     int64_t offset) {

        int i;
        for (i = 0 ; i < rktparlist->cnt ; i++)
		rktparlist->elems[i].offset = offset;
}


/**
 * Set offset values in partition list based on toppar's last stored offset.
 *
 *  from_rktp - true: set rktp's last stored offset, false: set def_value
 *  unless a concrete offset is set.
 *  is_commit: indicates that set offset is to be committed (for debug log)
 *
 * Returns the number of valid non-logical offsets (>=0).
 */
int rd_kafka_topic_partition_list_set_offsets (
	rd_kafka_t *rk,
        rd_kafka_topic_partition_list_t *rktparlist,
        int from_rktp, int64_t def_value, int is_commit) {
        int i;
	int valid_cnt = 0;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar = &rktparlist->elems[i];
		const char *verb = "setting";

                if (from_rktp) {
                        shptr_rd_kafka_toppar_t *s_rktp = rktpar->_private;
                        rd_kafka_toppar_t *rktp = rd_kafka_toppar_s2i(s_rktp);
                        rd_kafka_toppar_lock(rktp);

			rd_kafka_dbg(rk, CGRP | RD_KAFKA_DBG_TOPIC, "OFFSET",
				     "Topic %s [%"PRId32"]: "
				     "stored offset %"PRId64", committed "
				     "offset %"PRId64,
				     rktpar->topic, rktpar->partition,
				     rktp->rktp_stored_offset,
				     rktp->rktp_committed_offset);

			if (rktp->rktp_stored_offset >
			    rktp->rktp_committed_offset) {
				verb = "setting stored";
				rktpar->offset = rktp->rktp_stored_offset;
			} else {
				rktpar->offset = RD_KAFKA_OFFSET_INVALID;
			}
                        rd_kafka_toppar_unlock(rktp);
                } else {
			if (RD_KAFKA_OFFSET_IS_LOGICAL(rktpar->offset)) {
				verb = "setting default";
				rktpar->offset = def_value;
			} else
				verb = "keeping";
                }

		rd_kafka_dbg(rk, CGRP | RD_KAFKA_DBG_TOPIC, "OFFSET",
			     "Topic %s [%"PRId32"]: "
			     "%s offset %s%s",
			     rktpar->topic, rktpar->partition,
			     verb,
			     rd_kafka_offset2str(rktpar->offset),
			     is_commit ? " for commit" : "");

		if (!RD_KAFKA_OFFSET_IS_LOGICAL(rktpar->offset))
			valid_cnt++;
        }

	return valid_cnt;
}


/**
 * @returns the number of partitions with absolute (non-logical) offsets set.
 */
int rd_kafka_topic_partition_list_count_abs_offsets (
	const rd_kafka_topic_partition_list_t *rktparlist) {
	int i;
	int valid_cnt = 0;

        for (i = 0 ; i < rktparlist->cnt ; i++)
		if (!RD_KAFKA_OFFSET_IS_LOGICAL(rktparlist->elems[i].offset))
			valid_cnt++;

	return valid_cnt;
}

/**
 * @returns a new shared toppar pointer for partition at index 'idx',
 * or NULL if not set, not found, or out of range.
 *
 * @remark A new reference is returned.
 * @remark The _private field is set to the toppar it not previously set.
 */
shptr_rd_kafka_toppar_t *
rd_kafka_topic_partition_list_get_toppar (
        rd_kafka_t *rk, rd_kafka_topic_partition_t *rktpar) {
        shptr_rd_kafka_toppar_t *s_rktp;

        s_rktp = rd_kafka_topic_partition_get_toppar(rk, rktpar);
        if (!s_rktp)
                return NULL;

        return s_rktp;
}


/**
 * @brief Update _private (toppar) field to point to valid s_rktp
 *        for each parition.
 */
void
rd_kafka_topic_partition_list_update_toppars (rd_kafka_t *rk,
                                              rd_kafka_topic_partition_list_t
                                              *rktparlist) {
        int i;
        for (i = 0 ; i < rktparlist->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar = &rktparlist->elems[i];

                rd_kafka_topic_partition_list_get_toppar(rk, rktpar);
        }
}


/**
 * @brief Populate \p leaders with the leaders+partitions for the partitions in
 *        \p rktparlist. Duplicates are suppressed.
 *
 *        If no leader is found for a partition that element's \c .err will
 *        be set to RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE.
 *
 *        If the partition does not exist \c .err will be set to
 *        RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION.
 *
 * @param leaders rd_list_t of allocated (struct rd_kafka_partition_leader *)
 * @param query_topics (optional) rd_list of strdupped (char *)
 *
 * @remark This is based on the current topic_t and partition state
 *         which may lag behind the last metadata update due to internal
 *         threading and also the fact that no topic_t may have been created.
 *
 * @param leaders rd_list_t of type (struct rd_kafka_partition_leader *)
 *
 * @returns the number of leaders added.
 *
 * @sa rd_kafka_topic_partition_list_get_leaders_by_metadata
 *
 * @locks rd_kafka_*lock() MUST NOT be held
 */
int
rd_kafka_topic_partition_list_get_leaders (
        rd_kafka_t *rk,
        rd_kafka_topic_partition_list_t *rktparlist,
        rd_list_t *leaders,
        rd_list_t *query_topics) {
        int cnt = 0;
        int i;

        rd_kafka_rdlock(rk);

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar = &rktparlist->elems[i];
                rd_kafka_broker_t *rkb = NULL;
                struct rd_kafka_partition_leader leader_skel;
                struct rd_kafka_partition_leader *leader;
                const rd_kafka_metadata_topic_t *mtopic;
                const rd_kafka_metadata_partition_t *mpart;

                rd_kafka_metadata_cache_topic_partition_get(
                        rk, &mtopic, &mpart,
                        rktpar->topic, rktpar->partition, 1/*valid*/);

                if (mtopic &&
                    mtopic->err != RD_KAFKA_RESP_ERR_NO_ERROR &&
                    mtopic->err != RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE) {
                        /* Topic permanently errored */
                        rktpar->err = mtopic->err;
                        continue;
                }

                if (mtopic && !mpart && mtopic->partition_cnt > 0) {
                        /* Topic exists but partition doesnt.
                         * This is a permanent error. */
                        rktpar->err = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;
                        continue;
                }

                if (mpart &&
                    (mpart->leader == -1 ||
                     !(rkb = rd_kafka_broker_find_by_nodeid0(
                               rk, mpart->leader, -1/*any state*/)))) {
                        /* Partition has no (valid) leader */
                        rktpar->err =
                                mtopic->err ? mtopic->err :
                                RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE;
                }

                if (!mtopic || !rkb) {
                        /* Topic unknown or no current leader for partition,
                         * add topic to query list. */
                        if (query_topics &&
                            !rd_list_find(query_topics, rktpar->topic,
                                          (void *)strcmp))
                                rd_list_add(query_topics,
                                            rd_strdup(rktpar->topic));
                        continue;
                }

                /* Leader exists, add to leader list. */

                rktpar->err = RD_KAFKA_RESP_ERR_NO_ERROR;

                memset(&leader_skel, 0, sizeof(leader_skel));
                leader_skel.rkb = rkb;

                leader = rd_list_find(leaders, &leader_skel,
                                      rd_kafka_partition_leader_cmp);

                if (!leader) {
                        leader = rd_kafka_partition_leader_new(rkb);
                        rd_list_add(leaders, leader);
                        cnt++;
                }

                rd_kafka_topic_partition_copy(leader->partitions, rktpar);

                rd_kafka_broker_destroy(rkb);    /* loose refcount */
        }

        rd_kafka_rdunlock(rk);

        return cnt;

}




/**
 * @brief Get leaders for all partitions in \p rktparlist, querying metadata
 *        if needed.
 *
 * @param leaders is a pre-initialized (empty) list which will be populated
 *        with the leader brokers and their partitions
 *        (struct rd_kafka_partition_leader *)
 *
 * @returns an error code on error.
 *
 * @locks rd_kafka_*lock() MUST NOT be held
 */
rd_kafka_resp_err_t
rd_kafka_topic_partition_list_query_leaders (
        rd_kafka_t *rk,
        rd_kafka_topic_partition_list_t *rktparlist,
        rd_list_t *leaders, int timeout_ms) {
        rd_ts_t ts_end = rd_timeout_init(timeout_ms);
        rd_ts_t ts_query = 0;
        rd_ts_t now;
        int i = 0;

        /* Get all the partition leaders, try multiple times:
         * if there are no leaders after the first run fire off a leader
         * query and wait for broker state update before trying again,
         * keep trying and re-querying at increasing intervals until
         * success or timeout. */
        do {
                rd_list_t query_topics;
                int query_intvl;

                rd_list_init(&query_topics, rktparlist->cnt, rd_free);

                rd_kafka_topic_partition_list_get_leaders(
                        rk, rktparlist, leaders, &query_topics);

                if (rd_list_empty(&query_topics)) {
                        /* No remaining topics to query: leader-list complete.*/
                        rd_list_destroy(&query_topics);

                        /* No leader(s) for partitions means all partitions
                         * are unknown. */
                        if (rd_list_empty(leaders))
                                return RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;

                        return RD_KAFKA_RESP_ERR_NO_ERROR;
                }

                now = rd_clock();
                /*
                 * Missing leader for some partitions
                 */
                query_intvl = (i+1) * 100; /* add 100ms per iteration */
                if (query_intvl > 2*1000)
                        query_intvl = 2*1000; /* Cap to 2s */

                if (now >= ts_query + (query_intvl*1000)) {
                        /* Query metadata for missing leaders,
                         * possibly creating the topic. */
                        rd_kafka_metadata_refresh_topics(
                                rk, NULL, &query_topics, 1/*force*/,
                                "query partition leaders");
                        ts_query = now;
                } else {
                        /* Wait for broker ids to be updated from
                         * metadata refresh above. */
                        int wait_ms = rd_timeout_remains_limit(ts_end,
                                                               query_intvl);
                        rd_kafka_metadata_cache_wait_change(rk, wait_ms);
                }

                rd_list_destroy(&query_topics);

                i++;
        } while (ts_end == RD_POLL_INFINITE ||
                 now < ts_end); /* now is deliberately outdated here
                                 * since wait_change() will block.
                                 * This gives us one more chance to spin thru*/

        return RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE;
}


/**
 * @brief Populate \p rkts with the rd_kafka_itopic_t objects for the
 *        partitions in. Duplicates are suppressed.
 *
 * @returns the number of topics added.
 */
int
rd_kafka_topic_partition_list_get_topics (
        rd_kafka_t *rk,
        rd_kafka_topic_partition_list_t *rktparlist,
        rd_list_t *rkts) {
        int cnt = 0;

        int i;
        for (i = 0 ; i < rktparlist->cnt ; i++) {
                rd_kafka_topic_partition_t *rktpar = &rktparlist->elems[i];
                shptr_rd_kafka_toppar_t *s_rktp;
                rd_kafka_toppar_t *rktp;

                s_rktp = rd_kafka_topic_partition_get_toppar(rk, rktpar);
                if (!s_rktp) {
                        rktpar->err = RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION;
                        continue;
                }

                rktp = rd_kafka_toppar_s2i(s_rktp);

                if (!rd_list_find(rkts, rktp->rktp_s_rkt,
                                  rd_kafka_topic_cmp_s_rkt)) {
                        rd_list_add(rkts, rd_kafka_topic_keep(rktp->rktp_rkt));
                        cnt++;
                }

                rd_kafka_toppar_destroy(s_rktp);
        }

        return cnt;
}


/**
 * @brief Populate \p topics with the strdupped topic names in \p rktparlist.
 *        Duplicates are suppressed.
 *
 * @param include_regex: include regex topics
 *
 * @returns the number of topics added.
 */
int
rd_kafka_topic_partition_list_get_topic_names (
        const rd_kafka_topic_partition_list_t *rktparlist,
        rd_list_t *topics, int include_regex) {
        int cnt = 0;
        int i;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar = &rktparlist->elems[i];

                if (!include_regex && *rktpar->topic == '^')
                        continue;

                if (!rd_list_find(topics, rktpar->topic, (void *)strcmp)) {
                        rd_list_add(topics, rd_strdup(rktpar->topic));
                        cnt++;
                }
        }

        return cnt;
}


/**
 * @brief Create a copy of \p rktparlist only containing the partitions
 *        matched by \p match function.
 *
 * \p match shall return 1 for match, else 0.
 *
 * @returns a new list
 */
rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_match (
        const rd_kafka_topic_partition_list_t *rktparlist,
        int (*match) (const void *elem, const void *opaque),
        void *opaque) {
        rd_kafka_topic_partition_list_t *newlist;
        int i;

        newlist = rd_kafka_topic_partition_list_new(0);

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rktparlist->elems[i];

                if (!match(rktpar, opaque))
                        continue;

                rd_kafka_topic_partition_copy(newlist, rktpar);
        }

        return newlist;
}

void
rd_kafka_topic_partition_list_log (rd_kafka_t *rk, const char *fac, int dbg,
				   const rd_kafka_topic_partition_list_t *rktparlist) {
        int i;

	rd_kafka_dbg(rk, NONE|dbg, fac, "List with %d partition(s):",
		     rktparlist->cnt);
        for (i = 0 ; i < rktparlist->cnt ; i++) {
		const rd_kafka_topic_partition_t *rktpar =
			&rktparlist->elems[i];
		rd_kafka_dbg(rk, NONE|dbg, fac, " %s [%"PRId32"] offset %s%s%s",
			     rktpar->topic, rktpar->partition,
			     rd_kafka_offset2str(rktpar->offset),
			     rktpar->err ? ": error: " : "",
			     rktpar->err ? rd_kafka_err2str(rktpar->err) : "");
	}
}

/**
 * @returns a comma-separated list of partitions.
 */
const char *
rd_kafka_topic_partition_list_str (const rd_kafka_topic_partition_list_t *rktparlist,
                                   char *dest, size_t dest_size,
                                   int fmt_flags) {
        int i;
        size_t of = 0;
        int trunc = 0;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rktparlist->elems[i];
                char errstr[128];
                char offsetstr[32];
                int r;

                if (trunc) {
                        if (dest_size > 4)
                                rd_snprintf(&dest[dest_size-4], 4, "...");
                        break;
                }

                if (!rktpar->err && (fmt_flags & RD_KAFKA_FMT_F_ONLY_ERR))
                        continue;

                if (rktpar->err && !(fmt_flags & RD_KAFKA_FMT_F_NO_ERR))
                        rd_snprintf(errstr, sizeof(errstr),
                                    "(%s)", rd_kafka_err2str(rktpar->err));
                else
                        errstr[0] = '\0';

                if (rktpar->offset != RD_KAFKA_OFFSET_INVALID)
                        rd_snprintf(offsetstr, sizeof(offsetstr),
                                    "@%"PRId64, rktpar->offset);
                else
                        offsetstr[0] = '\0';

                r = rd_snprintf(&dest[of], dest_size-of,
                                "%s"
                                "%s[%"PRId32"]"
                                "%s"
                                "%s",
                                of == 0 ? "" : ", ",
                                rktpar->topic, rktpar->partition,
                                offsetstr,
                                errstr);

                if ((size_t)r >= dest_size-of)
                        trunc++;
                else
                        of += r;
        }

        return dest;
}



/**
 * @brief Update \p dst with info from \p src.
 *
 * Fields updated:
 *  - offset
 *  - err
 *
 * Will only partitions that are in both dst and src, other partitions will
 * remain unchanged.
 */
void
rd_kafka_topic_partition_list_update (rd_kafka_topic_partition_list_t *dst,
                                      const rd_kafka_topic_partition_list_t *src){
        int i;

        for (i = 0 ; i < dst->cnt ; i++) {
                rd_kafka_topic_partition_t *d = &dst->elems[i];
                rd_kafka_topic_partition_t *s;

                if (!(s = rd_kafka_topic_partition_list_find(
                              (rd_kafka_topic_partition_list_t *)src,
                              d->topic, d->partition)))
                        continue;

                d->offset = s->offset;
                d->err    = s->err;
        }
}


/**
 * @returns the sum of \p cb called for each element.
 */
size_t
rd_kafka_topic_partition_list_sum (
        const rd_kafka_topic_partition_list_t *rktparlist,
        size_t (*cb) (const rd_kafka_topic_partition_t *rktpar, void *opaque),
        void *opaque) {
        int i;
        size_t sum = 0;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rktparlist->elems[i];
                sum += cb(rktpar, opaque);
       }
        return sum;
}


/**
 * @brief Set \c .err field \p err on all partitions in list.
 */
void rd_kafka_topic_partition_list_set_err (
        rd_kafka_topic_partition_list_t *rktparlist,
        rd_kafka_resp_err_t err) {
        int i;

        for (i = 0 ; i < rktparlist->cnt ; i++)
                rktparlist->elems[i].err = err;
}


/**
 * @returns the number of wildcard/regex topics
 */
int rd_kafka_topic_partition_list_regex_cnt (
        const rd_kafka_topic_partition_list_t *rktparlist) {
        int i;
        int cnt = 0;

        for (i = 0 ; i < rktparlist->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rktparlist->elems[i];
                cnt += *rktpar->topic == '^';
        }
        return cnt;
}
