src/rdkafka_broker.c
void rd_kafka_broker_destroy_final (rd_kafka_broker_t *rkb) {

        rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_outbufs.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_waitresps.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_retrybufs.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_toppars));

        if (rkb->rkb_source != RD_KAFKA_INTERNAL &&
            (rkb->rkb_rk->rk_conf.security_protocol ==
             RD_KAFKA_PROTO_SASL_PLAINTEXT ||
             rkb->rkb_rk->rk_conf.security_protocol ==
             RD_KAFKA_PROTO_SASL_SSL))
                rd_kafka_sasl_broker_term(rkb);

        if (rkb->rkb_wakeup_fd[0] != -1)
                rd_close(rkb->rkb_wakeup_fd[0]);
        if (rkb->rkb_wakeup_fd[1] != -1)
                rd_close(rkb->rkb_wakeup_fd[1]);

	if (rkb->rkb_recv_buf)
		rd_kafka_buf_destroy(rkb->rkb_recv_buf);

	if (rkb->rkb_rsal)
		rd_sockaddr_list_destroy(rkb->rkb_rsal);

	if (rkb->rkb_ApiVersions)
		rd_free(rkb->rkb_ApiVersions);
        rd_free(rkb->rkb_origname);

	rd_kafka_q_purge(rkb->rkb_ops);
        rd_kafka_q_destroy_owner(rkb->rkb_ops);

        rd_avg_destroy(&rkb->rkb_avg_int_latency);
        rd_avg_destroy(&rkb->rkb_avg_outbuf_latency);
        rd_avg_destroy(&rkb->rkb_avg_rtt);
	rd_avg_destroy(&rkb->rkb_avg_throttle);

        mtx_lock(&rkb->rkb_logname_lock);
        rd_free(rkb->rkb_logname);
        rkb->rkb_logname = NULL;
        mtx_unlock(&rkb->rkb_logname_lock);
        mtx_destroy(&rkb->rkb_logname_lock);

	mtx_destroy(&rkb->rkb_lock);

        rd_refcnt_destroy(&rkb->rkb_refcnt);

	rd_free(rkb);
}
/**
 * Final destructor. Refcnt must be 0.
 */
Final destructor. Refcnt must be 0.
Final destructor . Refcnt must be 0 .
void rd_kafka_broker_destroy_final (rd_kafka_broker_t *rkb) {
        rd_kafka_assert(rkb->rkb_rk, thrd_is_current(rkb->rkb_thread));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_outbufs.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_waitresps.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_retrybufs.rkbq_bufs));
        rd_kafka_assert(rkb->rkb_rk, TAILQ_EMPTY(&rkb->rkb_toppars));
        if (rkb->rkb_source != RD_KAFKA_INTERNAL &&
            (rkb->rkb_rk->rk_conf.security_protocol ==
             RD_KAFKA_PROTO_SASL_PLAINTEXT ||
             rkb->rkb_rk->rk_conf.security_protocol ==
             RD_KAFKA_PROTO_SASL_SSL))
                rd_kafka_sasl_broker_term(rkb);
        if (rkb->rkb_wakeup_fd[0] != -1)
                rd_close(rkb->rkb_wakeup_fd[0]);
        if (rkb->rkb_wakeup_fd[1] != -1)
                rd_close(rkb->rkb_wakeup_fd[1]);
	if (rkb->rkb_recv_buf)
		rd_kafka_buf_destroy(rkb->rkb_recv_buf);
	if (rkb->rkb_rsal)
		rd_sockaddr_list_destroy(rkb->rkb_rsal);
	if (rkb->rkb_ApiVersions)
		rd_free(rkb->rkb_ApiVersions);
        rd_free(rkb->rkb_origname);
	rd_kafka_q_purge(rkb->rkb_ops);
        rd_kafka_q_destroy_owner(rkb->rkb_ops);
        rd_avg_destroy(&rkb->rkb_avg_int_latency);
        rd_avg_destroy(&rkb->rkb_avg_outbuf_latency);
        rd_avg_destroy(&rkb->rkb_avg_rtt);
	rd_avg_destroy(&rkb->rkb_avg_throttle);
        mtx_lock(&rkb->rkb_logname_lock);
        rd_free(rkb->rkb_logname);
        rkb->rkb_logname = NULL;
        mtx_unlock(&rkb->rkb_logname_lock);
        mtx_destroy(&rkb->rkb_logname_lock);
	mtx_destroy(&rkb->rkb_lock);
        rd_refcnt_destroy(&rkb->rkb_refcnt);
	rd_free(rkb);
}
void rd_kafka_broker_destroy_final ( rd_kafka_broker_t * rkb ) { rd_kafka_assert ( rkb -> rkb_rk , thrd_is_current ( rkb -> rkb_thread ) ) ; rd_kafka_assert ( rkb -> rkb_rk , TAILQ_EMPTY ( & rkb -> rkb_outbufs . rkbq_bufs ) ) ; rd_kafka_assert ( rkb -> rkb_rk , TAILQ_EMPTY ( & rkb -> rkb_waitresps . rkbq_bufs ) ) ; rd_kafka_assert ( rkb -> rkb_rk , TAILQ_EMPTY ( & rkb -> rkb_retrybufs . rkbq_bufs ) ) ; rd_kafka_assert ( rkb -> rkb_rk , TAILQ_EMPTY ( & rkb -> rkb_toppars ) ) ; if ( rkb -> rkb_source != RD_KAFKA_INTERNAL && ( rkb -> rkb_rk -> rk_conf . security_protocol == RD_KAFKA_PROTO_SASL_PLAINTEXT || rkb -> rkb_rk -> rk_conf . security_protocol == RD_KAFKA_PROTO_SASL_SSL ) ) rd_kafka_sasl_broker_term ( rkb ) ; if ( rkb -> rkb_wakeup_fd [ 0 ] != -1 ) rd_close ( rkb -> rkb_wakeup_fd [ 0 ] ) ; if ( rkb -> rkb_wakeup_fd [ 1 ] != -1 ) rd_close ( rkb -> rkb_wakeup_fd [ 1 ] ) ; if ( rkb -> rkb_recv_buf ) rd_kafka_buf_destroy ( rkb -> rkb_recv_buf ) ; if ( rkb -> rkb_rsal ) rd_sockaddr_list_destroy ( rkb -> rkb_rsal ) ; if ( rkb -> rkb_ApiVersions ) rd_free ( rkb -> rkb_ApiVersions ) ; rd_free ( rkb -> rkb_origname ) ; rd_kafka_q_purge ( rkb -> rkb_ops ) ; rd_kafka_q_destroy_owner ( rkb -> rkb_ops ) ; rd_avg_destroy ( & rkb -> rkb_avg_int_latency ) ; rd_avg_destroy ( & rkb -> rkb_avg_outbuf_latency ) ; rd_avg_destroy ( & rkb -> rkb_avg_rtt ) ; rd_avg_destroy ( & rkb -> rkb_avg_throttle ) ; mtx_lock ( & rkb -> rkb_logname_lock ) ; rd_free ( rkb -> rkb_logname ) ; rkb -> rkb_logname = NULL ; mtx_unlock ( & rkb -> rkb_logname_lock ) ; mtx_destroy ( & rkb -> rkb_logname_lock ) ; mtx_destroy ( & rkb -> rkb_lock ) ; rd_refcnt_destroy ( & rkb -> rkb_refcnt ) ; rd_free ( rkb ) ; }
Final destructor.
Final destructor .