src/rdkafka.c
void rd_kafka_destroy_final (rd_kafka_t *rk) {

        rd_kafka_assert(rk, rd_kafka_terminating(rk));

        /* Synchronize state */
        rd_kafka_wrlock(rk);
        rd_kafka_wrunlock(rk);

        rd_kafka_assignors_term(rk);

        rd_kafka_metadata_cache_destroy(rk);

        /* Terminate SASL provider */
        if (rk->rk_conf.sasl.provider)
                rd_kafka_sasl_term(rk);

        rd_kafka_timers_destroy(&rk->rk_timers);

        rd_kafka_dbg(rk, GENERIC, "TERMINATE", "Destroying op queues");

        /* Destroy cgrp */
        if (rk->rk_cgrp) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                             "Destroying cgrp");
                /* Reset queue forwarding (rep -> cgrp) */
                rd_kafka_q_fwd_set(rk->rk_rep, NULL);
                rd_kafka_cgrp_destroy_final(rk->rk_cgrp);
        }

	/* Purge op-queues */
	rd_kafka_q_destroy_owner(rk->rk_rep);
	rd_kafka_q_destroy_owner(rk->rk_ops);

#if WITH_SSL
        if (rk->rk_conf.ssl.ctx) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE", "Destroying SSL CTX");
                rd_kafka_ssl_ctx_term(rk);
        }
#endif

        /* It is not safe to log after this point. */
        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Termination done: freeing resources");

        if (rk->rk_logq) {
                rd_kafka_q_destroy_owner(rk->rk_logq);
                rk->rk_logq = NULL;
        }

        if (rk->rk_type == RD_KAFKA_PRODUCER) {
		cnd_destroy(&rk->rk_curr_msgs.cnd);
		mtx_destroy(&rk->rk_curr_msgs.lock);
	}

        if (rk->rk_fatal.errstr) {
                rd_free(rk->rk_fatal.errstr);
                rk->rk_fatal.errstr = NULL;
        }

	cnd_destroy(&rk->rk_broker_state_change_cnd);
	mtx_destroy(&rk->rk_broker_state_change_lock);

        mtx_destroy(&rk->rk_suppress.sparse_connect_lock);

        cnd_destroy(&rk->rk_init_cnd);
        mtx_destroy(&rk->rk_init_lock);

	if (rk->rk_full_metadata)
		rd_kafka_metadata_destroy(rk->rk_full_metadata);
        rd_kafkap_str_destroy(rk->rk_client_id);
        rd_kafkap_str_destroy(rk->rk_group_id);
        rd_kafkap_str_destroy(rk->rk_eos.transactional_id);
	rd_kafka_anyconf_destroy(_RK_GLOBAL, &rk->rk_conf);
        rd_list_destroy(&rk->rk_broker_by_id);

	rd_kafkap_bytes_destroy((rd_kafkap_bytes_t *)rk->rk_null_bytes);
	rwlock_destroy(&rk->rk_lock);

	rd_free(rk);
	rd_kafka_global_cnt_decr();
}
/**
 * @brief Final destructor for rd_kafka_t, must only be called with refcnt 0.
 *
 * @locality application thread
 */
@brief Final destructor for rd_kafka_t, must only be called with refcnt 0.
@locality application thread
@brief Final destructor for rd_kafka_t must only be called with refcnt 0 . @locality application thread
void rd_kafka_destroy_final (rd_kafka_t *rk) {
        rd_kafka_assert(rk, rd_kafka_terminating(rk));
        rd_kafka_wrlock(rk);
        rd_kafka_wrunlock(rk);
        rd_kafka_assignors_term(rk);
        rd_kafka_metadata_cache_destroy(rk);
        if (rk->rk_conf.sasl.provider)
                rd_kafka_sasl_term(rk);
        rd_kafka_timers_destroy(&rk->rk_timers);
        rd_kafka_dbg(rk, GENERIC, "TERMINATE", "Destroying op queues");
        if (rk->rk_cgrp) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                             "Destroying cgrp");
                rd_kafka_q_fwd_set(rk->rk_rep, NULL);
                rd_kafka_cgrp_destroy_final(rk->rk_cgrp);
        }
	rd_kafka_q_destroy_owner(rk->rk_rep);
	rd_kafka_q_destroy_owner(rk->rk_ops);
#if WITH_SSL
        if (rk->rk_conf.ssl.ctx) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE", "Destroying SSL CTX");
                rd_kafka_ssl_ctx_term(rk);
        }
#endif
        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Termination done: freeing resources");
        if (rk->rk_logq) {
                rd_kafka_q_destroy_owner(rk->rk_logq);
                rk->rk_logq = NULL;
        }
        if (rk->rk_type == RD_KAFKA_PRODUCER) {
		cnd_destroy(&rk->rk_curr_msgs.cnd);
		mtx_destroy(&rk->rk_curr_msgs.lock);
	}
        if (rk->rk_fatal.errstr) {
                rd_free(rk->rk_fatal.errstr);
                rk->rk_fatal.errstr = NULL;
        }
	cnd_destroy(&rk->rk_broker_state_change_cnd);
	mtx_destroy(&rk->rk_broker_state_change_lock);
        mtx_destroy(&rk->rk_suppress.sparse_connect_lock);
        cnd_destroy(&rk->rk_init_cnd);
        mtx_destroy(&rk->rk_init_lock);
	if (rk->rk_full_metadata)
		rd_kafka_metadata_destroy(rk->rk_full_metadata);
        rd_kafkap_str_destroy(rk->rk_client_id);
        rd_kafkap_str_destroy(rk->rk_group_id);
        rd_kafkap_str_destroy(rk->rk_eos.transactional_id);
	rd_kafka_anyconf_destroy(_RK_GLOBAL, &rk->rk_conf);
        rd_list_destroy(&rk->rk_broker_by_id);
	rd_kafkap_bytes_destroy((rd_kafkap_bytes_t *)rk->rk_null_bytes);
	rwlock_destroy(&rk->rk_lock);
	rd_free(rk);
	rd_kafka_global_cnt_decr();
}
void rd_kafka_destroy_final ( rd_kafka_t * rk ) { rd_kafka_assert ( rk , rd_kafka_terminating ( rk ) ) ; rd_kafka_wrlock ( rk ) ; rd_kafka_wrunlock ( rk ) ; rd_kafka_assignors_term ( rk ) ; rd_kafka_metadata_cache_destroy ( rk ) ; if ( rk -> rk_conf . sasl . provider ) rd_kafka_sasl_term ( rk ) ; rd_kafka_timers_destroy ( & rk -> rk_timers ) ; rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; if ( rk -> rk_cgrp ) { rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; rd_kafka_q_fwd_set ( rk -> rk_rep , NULL ) ; rd_kafka_cgrp_destroy_final ( rk -> rk_cgrp ) ; } rd_kafka_q_destroy_owner ( rk -> rk_rep ) ; rd_kafka_q_destroy_owner ( rk -> rk_ops ) ; #if WITH_SSL if ( rk -> rk_conf . ssl . ctx ) { rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; rd_kafka_ssl_ctx_term ( rk ) ; } #endif rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; if ( rk -> rk_logq ) { rd_kafka_q_destroy_owner ( rk -> rk_logq ) ; rk -> rk_logq = NULL ; } if ( rk -> rk_type == RD_KAFKA_PRODUCER ) { cnd_destroy ( & rk -> rk_curr_msgs . cnd ) ; mtx_destroy ( & rk -> rk_curr_msgs . lock ) ; } if ( rk -> rk_fatal . errstr ) { rd_free ( rk -> rk_fatal . errstr ) ; rk -> rk_fatal . errstr = NULL ; } cnd_destroy ( & rk -> rk_broker_state_change_cnd ) ; mtx_destroy ( & rk -> rk_broker_state_change_lock ) ; mtx_destroy ( & rk -> rk_suppress . sparse_connect_lock ) ; cnd_destroy ( & rk -> rk_init_cnd ) ; mtx_destroy ( & rk -> rk_init_lock ) ; if ( rk -> rk_full_metadata ) rd_kafka_metadata_destroy ( rk -> rk_full_metadata ) ; rd_kafkap_str_destroy ( rk -> rk_client_id ) ; rd_kafkap_str_destroy ( rk -> rk_group_id ) ; rd_kafkap_str_destroy ( rk -> rk_eos . transactional_id ) ; rd_kafka_anyconf_destroy ( _RK_GLOBAL , & rk -> rk_conf ) ; rd_list_destroy ( & rk -> rk_broker_by_id ) ; rd_kafkap_bytes_destroy ( ( rd_kafkap_bytes_t * ) rk -> rk_null_bytes ) ; rwlock_destroy ( & rk -> rk_lock ) ; rd_free ( rk ) ; rd_kafka_global_cnt_decr ( ) ; }
@brief Final destructor for rd_kafka_t, must only be called with refcnt 0.
@brief Final destructor for rd_kafka_t must only be called with refcnt 0 .
/* Synchronize state */ /* Terminate SASL provider */ /* Destroy cgrp */ /* Reset queue forwarding (rep -> cgrp) */ /* Purge op-queues */ /* It is not safe to log after this point. */