src/rdkafka_broker.c
static void rd_kafka_broker_handle_purge_queues (rd_kafka_broker_t *rkb,
                                                 rd_kafka_op_t *rko) {
        int purge_flags = rko->rko_u.purge.flags;
        int inflight_cnt = 0, retry_cnt = 0, outq_cnt = 0, partial_cnt = 0;

        rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGE",
                   "Purging queues with flags %s",
                   rd_kafka_purge_flags2str(purge_flags));


        /**
         * First purge any Produce requests to move the
         * messages from the request's message queue to delivery reports.
         */

        /* Purge in-flight ProduceRequests */
        if (purge_flags & RD_KAFKA_PURGE_F_INFLIGHT)
                inflight_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 1, &rkb->rkb_waitresps, NULL, RD_KAFKAP_Produce,
                        RD_KAFKA_RESP_ERR__PURGE_INFLIGHT, 0, NULL, 0);

        if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) {
                /* Requests in retry queue */
                retry_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 0, &rkb->rkb_retrybufs, NULL, RD_KAFKAP_Produce,
                        RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0, NULL, 0);

                /* Requests in transmit queue not completely sent yet.
                 * partial_cnt is included in outq_cnt and denotes a request
                 * that has been partially transmitted. */
                outq_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 0, &rkb->rkb_outbufs, &partial_cnt,
                        RD_KAFKAP_Produce, RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0,
                        NULL, 0);

                /* Purging a partially transmitted request will mess up
                 * the protocol stream, so we need to disconnect from the broker
                 * to get a clean protocol socket. */
                if (partial_cnt)
                        rd_kafka_broker_fail(
                                rkb, LOG_NOTICE,
                                RD_KAFKA_RESP_ERR__PURGE_QUEUE,
                                "purged %d partially sent request: "
                                "forcing disconnect", partial_cnt);
        }

        rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ",
                   "Purged %i in-flight, %i retry-queued, "
                   "%i out-queue, %i partially-sent requests",
                   inflight_cnt, retry_cnt, outq_cnt, partial_cnt);

        /* Purge partition queues */
        if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) {
                rd_kafka_toppar_t *rktp;
                int msg_cnt = 0;
                int part_cnt = 0;

                TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) {
                        int r;

                        r = rd_kafka_toppar_handle_purge_queues(rktp, rkb,
                                                                purge_flags);
                        if (r > 0) {
                                msg_cnt += r;
                                part_cnt++;
                        }
                }

                rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ",
                           "Purged %i message(s) from %d partition(s)",
                           msg_cnt, part_cnt);
        }

        rd_kafka_op_reply(rko, RD_KAFKA_RESP_ERR_NO_ERROR);
}
/**
 * @brief Handle purge queues request
 *
 * @locality broker thread
 * @locks none
 */
@brief Handle purge queues request
@locality broker thread
@locks none
@brief Handle purge queues request @locality broker thread @locks none
static void rd_kafka_broker_handle_purge_queues (rd_kafka_broker_t *rkb,
                                                 rd_kafka_op_t *rko) {
        int purge_flags = rko->rko_u.purge.flags;
        int inflight_cnt = 0, retry_cnt = 0, outq_cnt = 0, partial_cnt = 0;
        rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGE",
                   "Purging queues with flags %s",
                   rd_kafka_purge_flags2str(purge_flags));
        if (purge_flags & RD_KAFKA_PURGE_F_INFLIGHT)
                inflight_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 1, &rkb->rkb_waitresps, NULL, RD_KAFKAP_Produce,
                        RD_KAFKA_RESP_ERR__PURGE_INFLIGHT, 0, NULL, 0);
        if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) {
                retry_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 0, &rkb->rkb_retrybufs, NULL, RD_KAFKAP_Produce,
                        RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0, NULL, 0);
                outq_cnt = rd_kafka_broker_bufq_timeout_scan(
                        rkb, 0, &rkb->rkb_outbufs, &partial_cnt,
                        RD_KAFKAP_Produce, RD_KAFKA_RESP_ERR__PURGE_QUEUE, 0,
                        NULL, 0);
                if (partial_cnt)
                        rd_kafka_broker_fail(
                                rkb, LOG_NOTICE,
                                RD_KAFKA_RESP_ERR__PURGE_QUEUE,
                                "purged %d partially sent request: "
                                "forcing disconnect", partial_cnt);
        }
        rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ",
                   "Purged %i in-flight, %i retry-queued, "
                   "%i out-queue, %i partially-sent requests",
                   inflight_cnt, retry_cnt, outq_cnt, partial_cnt);
        if (purge_flags & RD_KAFKA_PURGE_F_QUEUE) {
                rd_kafka_toppar_t *rktp;
                int msg_cnt = 0;
                int part_cnt = 0;
                TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) {
                        int r;
                        r = rd_kafka_toppar_handle_purge_queues(rktp, rkb,
                                                                purge_flags);
                        if (r > 0) {
                                msg_cnt += r;
                                part_cnt++;
                        }
                }
                rd_rkb_dbg(rkb, QUEUE|RD_KAFKA_DBG_TOPIC, "PURGEQ",
                           "Purged %i message(s) from %d partition(s)",
                           msg_cnt, part_cnt);
        }
        rd_kafka_op_reply(rko, RD_KAFKA_RESP_ERR_NO_ERROR);
}
static void rd_kafka_broker_handle_purge_queues ( rd_kafka_broker_t * rkb , rd_kafka_op_t * rko ) { int purge_flags = rko -> rko_u . purge . flags ; int inflight_cnt = 0 , retry_cnt = 0 , outq_cnt = 0 , partial_cnt = 0 ; rd_rkb_dbg ( rkb , QUEUE | RD_KAFKA_DBG_TOPIC , " " , " " , rd_kafka_purge_flags2str ( purge_flags ) ) ; if ( purge_flags & RD_KAFKA_PURGE_F_INFLIGHT ) inflight_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 1 , & rkb -> rkb_waitresps , NULL , RD_KAFKAP_Produce , RD_KAFKA_RESP_ERR__PURGE_INFLIGHT , 0 , NULL , 0 ) ; if ( purge_flags & RD_KAFKA_PURGE_F_QUEUE ) { retry_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 0 , & rkb -> rkb_retrybufs , NULL , RD_KAFKAP_Produce , RD_KAFKA_RESP_ERR__PURGE_QUEUE , 0 , NULL , 0 ) ; outq_cnt = rd_kafka_broker_bufq_timeout_scan ( rkb , 0 , & rkb -> rkb_outbufs , & partial_cnt , RD_KAFKAP_Produce , RD_KAFKA_RESP_ERR__PURGE_QUEUE , 0 , NULL , 0 ) ; if ( partial_cnt ) rd_kafka_broker_fail ( rkb , LOG_NOTICE , RD_KAFKA_RESP_ERR__PURGE_QUEUE , " " " " , partial_cnt ) ; } rd_rkb_dbg ( rkb , QUEUE | RD_KAFKA_DBG_TOPIC , " " , " " " " , inflight_cnt , retry_cnt , outq_cnt , partial_cnt ) ; if ( purge_flags & RD_KAFKA_PURGE_F_QUEUE ) { rd_kafka_toppar_t * rktp ; int msg_cnt = 0 ; int part_cnt = 0 ; TAILQ_FOREACH ( rktp , & rkb -> rkb_toppars , rktp_rkblink ) { int r ; r = rd_kafka_toppar_handle_purge_queues ( rktp , rkb , purge_flags ) ; if ( r > 0 ) { msg_cnt += r ; part_cnt ++ ; } } rd_rkb_dbg ( rkb , QUEUE | RD_KAFKA_DBG_TOPIC , " " , " " , msg_cnt , part_cnt ) ; } rd_kafka_op_reply ( rko , RD_KAFKA_RESP_ERR_NO_ERROR ) ; }
@brief Handle purge queues request
 @locality broker thread
 @locks none
@brief Handle purge queues request @locality broker thread @locks none
/**
         * First purge any Produce requests to move the
         * messages from the request's message queue to delivery reports.
         */ /* Purge in-flight ProduceRequests */ /* Requests in retry queue */ /* Requests in transmit queue not completely sent yet.
                 * partial_cnt is included in outq_cnt and denotes a request
                 * that has been partially transmitted. */ /* Purging a partially transmitted request will mess up
                 * the protocol stream, so we need to disconnect from the broker
                 * to get a clean protocol socket. */ /* Purge partition queues */