src/rdkafka.c
static void rd_kafka_destroy_internal (rd_kafka_t *rk) {
	rd_kafka_itopic_t *rkt, *rkt_tmp;
	rd_kafka_broker_t *rkb, *rkb_tmp;
        rd_list_t wait_thrds;
        thrd_t *thrd;
        int i;

        rd_kafka_dbg(rk, ALL, "DESTROY", "Destroy internal");

        /* Trigger any state-change waiters (which should check the
         * terminate flag whenever they wake up). */
        rd_kafka_brokers_broadcast_state_change(rk);

        if (rk->rk_background.thread) {
                int res;
                /* Send op to trigger queue/io wake-up.
                 * The op itself is (likely) ignored by the receiver. */
                rd_kafka_q_enq(rk->rk_background.q,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));

                rd_kafka_dbg(rk, ALL, "DESTROY",
                             "Waiting for background queue thread "
                             "to terminate");
                thrd_join(rk->rk_background.thread, &res);
                rd_kafka_q_destroy_owner(rk->rk_background.q);
        }

        /* Call on_destroy() interceptors */
        rd_kafka_interceptors_on_destroy(rk);

	/* Brokers pick up on rk_terminate automatically. */

        /* List of (broker) threads to join to synchronize termination */
        rd_list_init(&wait_thrds, rd_atomic32_get(&rk->rk_broker_cnt), NULL);

	rd_kafka_wrlock(rk);

        rd_kafka_dbg(rk, ALL, "DESTROY", "Removing all topics");
	/* Decommission all topics */
	TAILQ_FOREACH_SAFE(rkt, &rk->rk_topics, rkt_link, rkt_tmp) {
		rd_kafka_wrunlock(rk);
		rd_kafka_topic_partitions_remove(rkt);
		rd_kafka_wrlock(rk);
	}

        /* Decommission brokers.
         * Broker thread holds a refcount and detects when broker refcounts
         * reaches 1 and then decommissions itself. */
        TAILQ_FOREACH_SAFE(rkb, &rk->rk_brokers, rkb_link, rkb_tmp) {
                /* Add broker's thread to wait_thrds list for later joining */
                thrd = malloc(sizeof(*thrd));
                *thrd = rkb->rkb_thread;
                rd_list_add(&wait_thrds, thrd);
                rd_kafka_wrunlock(rk);

                rd_kafka_dbg(rk, BROKER, "DESTROY",
                             "Sending TERMINATE to %s",
                             rd_kafka_broker_name(rkb));
                /* Send op to trigger queue/io wake-up.
                 * The op itself is (likely) ignored by the broker thread. */
                rd_kafka_q_enq(rkb->rkb_ops,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));

#ifndef _MSC_VER
                /* Interrupt IO threads to speed up termination. */
                if (rk->rk_conf.term_sig)
			pthread_kill(rkb->rkb_thread, rk->rk_conf.term_sig);
#endif

                rd_kafka_broker_destroy(rkb);

                rd_kafka_wrlock(rk);
        }

        if (rk->rk_clusterid) {
                rd_free(rk->rk_clusterid);
                rk->rk_clusterid = NULL;
        }

        rd_kafka_wrunlock(rk);

        mtx_lock(&rk->rk_broker_state_change_lock);
        /* Purge broker state change waiters */
        rd_list_destroy(&rk->rk_broker_state_change_waiters);
        mtx_unlock(&rk->rk_broker_state_change_lock);

        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Purging reply queue");

	/* Purge op-queue */
        rd_kafka_q_disable(rk->rk_rep);
	rd_kafka_q_purge(rk->rk_rep);

	/* Loose our special reference to the internal broker. */
        mtx_lock(&rk->rk_internal_rkb_lock);
	if ((rkb = rk->rk_internal_rkb)) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                             "Decommissioning internal broker");

                /* Send op to trigger queue wake-up. */
                rd_kafka_q_enq(rkb->rkb_ops,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));

                rk->rk_internal_rkb = NULL;
                thrd = malloc(sizeof(*thrd));
                *thrd = rkb->rkb_thread;
                rd_list_add(&wait_thrds, thrd);
        }
        mtx_unlock(&rk->rk_internal_rkb_lock);
	if (rkb)
		rd_kafka_broker_destroy(rkb);


        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Join %d broker thread(s)", rd_list_cnt(&wait_thrds));

        /* Join broker threads */
        RD_LIST_FOREACH(thrd, &wait_thrds, i) {
                int res;
                if (thrd_join(*thrd, &res) != thrd_success)
                        ;
                free(thrd);
        }

        rd_list_destroy(&wait_thrds);
}
/**
 * Main destructor for rd_kafka_t
 *
 * Locality: rdkafka main thread or application thread during rd_kafka_new()
 */
Main destructor for rd_kafka_t
Locality: rdkafka main thread or application thread during rd_kafka_new()
Main destructor for rd_kafka_t Locality : rdkafka main thread or application thread during rd_kafka_new ()
static void rd_kafka_destroy_internal (rd_kafka_t *rk) {
	rd_kafka_itopic_t *rkt, *rkt_tmp;
	rd_kafka_broker_t *rkb, *rkb_tmp;
        rd_list_t wait_thrds;
        thrd_t *thrd;
        int i;
        rd_kafka_dbg(rk, ALL, "DESTROY", "Destroy internal");
        rd_kafka_brokers_broadcast_state_change(rk);
        if (rk->rk_background.thread) {
                int res;
                rd_kafka_q_enq(rk->rk_background.q,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));
                rd_kafka_dbg(rk, ALL, "DESTROY",
                             "Waiting for background queue thread "
                             "to terminate");
                thrd_join(rk->rk_background.thread, &res);
                rd_kafka_q_destroy_owner(rk->rk_background.q);
        }
        rd_kafka_interceptors_on_destroy(rk);
        rd_list_init(&wait_thrds, rd_atomic32_get(&rk->rk_broker_cnt), NULL);
	rd_kafka_wrlock(rk);
        rd_kafka_dbg(rk, ALL, "DESTROY", "Removing all topics");
	TAILQ_FOREACH_SAFE(rkt, &rk->rk_topics, rkt_link, rkt_tmp) {
		rd_kafka_wrunlock(rk);
		rd_kafka_topic_partitions_remove(rkt);
		rd_kafka_wrlock(rk);
	}
        TAILQ_FOREACH_SAFE(rkb, &rk->rk_brokers, rkb_link, rkb_tmp) {
                thrd = malloc(sizeof(*thrd));
                *thrd = rkb->rkb_thread;
                rd_list_add(&wait_thrds, thrd);
                rd_kafka_wrunlock(rk);
                rd_kafka_dbg(rk, BROKER, "DESTROY",
                             "Sending TERMINATE to %s",
                             rd_kafka_broker_name(rkb));
                rd_kafka_q_enq(rkb->rkb_ops,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));
#ifndef _MSC_VER
                if (rk->rk_conf.term_sig)
			pthread_kill(rkb->rkb_thread, rk->rk_conf.term_sig);
#endif
                rd_kafka_broker_destroy(rkb);
                rd_kafka_wrlock(rk);
        }
        if (rk->rk_clusterid) {
                rd_free(rk->rk_clusterid);
                rk->rk_clusterid = NULL;
        }
        rd_kafka_wrunlock(rk);
        mtx_lock(&rk->rk_broker_state_change_lock);
        rd_list_destroy(&rk->rk_broker_state_change_waiters);
        mtx_unlock(&rk->rk_broker_state_change_lock);
        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Purging reply queue");
        rd_kafka_q_disable(rk->rk_rep);
	rd_kafka_q_purge(rk->rk_rep);
        mtx_lock(&rk->rk_internal_rkb_lock);
	if ((rkb = rk->rk_internal_rkb)) {
                rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                             "Decommissioning internal broker");
                rd_kafka_q_enq(rkb->rkb_ops,
                               rd_kafka_op_new(RD_KAFKA_OP_TERMINATE));
                rk->rk_internal_rkb = NULL;
                thrd = malloc(sizeof(*thrd));
                *thrd = rkb->rkb_thread;
                rd_list_add(&wait_thrds, thrd);
        }
        mtx_unlock(&rk->rk_internal_rkb_lock);
	if (rkb)
		rd_kafka_broker_destroy(rkb);
        rd_kafka_dbg(rk, GENERIC, "TERMINATE",
                     "Join %d broker thread(s)", rd_list_cnt(&wait_thrds));
        RD_LIST_FOREACH(thrd, &wait_thrds, i) {
                int res;
                if (thrd_join(*thrd, &res) != thrd_success)
                        ;
                free(thrd);
        }
        rd_list_destroy(&wait_thrds);
}
static void rd_kafka_destroy_internal ( rd_kafka_t * rk ) { rd_kafka_itopic_t * rkt , * rkt_tmp ; rd_kafka_broker_t * rkb , * rkb_tmp ; rd_list_t wait_thrds ; thrd_t * thrd ; int i ; rd_kafka_dbg ( rk , ALL , " " , " " ) ; rd_kafka_brokers_broadcast_state_change ( rk ) ; if ( rk -> rk_background . thread ) { int res ; rd_kafka_q_enq ( rk -> rk_background . q , rd_kafka_op_new ( RD_KAFKA_OP_TERMINATE ) ) ; rd_kafka_dbg ( rk , ALL , " " , " " " " ) ; thrd_join ( rk -> rk_background . thread , & res ) ; rd_kafka_q_destroy_owner ( rk -> rk_background . q ) ; } rd_kafka_interceptors_on_destroy ( rk ) ; rd_list_init ( & wait_thrds , rd_atomic32_get ( & rk -> rk_broker_cnt ) , NULL ) ; rd_kafka_wrlock ( rk ) ; rd_kafka_dbg ( rk , ALL , " " , " " ) ; TAILQ_FOREACH_SAFE ( rkt , & rk -> rk_topics , rkt_link , rkt_tmp ) { rd_kafka_wrunlock ( rk ) ; rd_kafka_topic_partitions_remove ( rkt ) ; rd_kafka_wrlock ( rk ) ; } TAILQ_FOREACH_SAFE ( rkb , & rk -> rk_brokers , rkb_link , rkb_tmp ) { thrd = malloc ( sizeof ( * thrd ) ) ; * thrd = rkb -> rkb_thread ; rd_list_add ( & wait_thrds , thrd ) ; rd_kafka_wrunlock ( rk ) ; rd_kafka_dbg ( rk , BROKER , " " , " " , rd_kafka_broker_name ( rkb ) ) ; rd_kafka_q_enq ( rkb -> rkb_ops , rd_kafka_op_new ( RD_KAFKA_OP_TERMINATE ) ) ; #ifndef _MSC_VER if ( rk -> rk_conf . term_sig ) pthread_kill ( rkb -> rkb_thread , rk -> rk_conf . term_sig ) ; #endif rd_kafka_broker_destroy ( rkb ) ; rd_kafka_wrlock ( rk ) ; } if ( rk -> rk_clusterid ) { rd_free ( rk -> rk_clusterid ) ; rk -> rk_clusterid = NULL ; } rd_kafka_wrunlock ( rk ) ; mtx_lock ( & rk -> rk_broker_state_change_lock ) ; rd_list_destroy ( & rk -> rk_broker_state_change_waiters ) ; mtx_unlock ( & rk -> rk_broker_state_change_lock ) ; rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; rd_kafka_q_disable ( rk -> rk_rep ) ; rd_kafka_q_purge ( rk -> rk_rep ) ; mtx_lock ( & rk -> rk_internal_rkb_lock ) ; if ( ( rkb = rk -> rk_internal_rkb ) ) { rd_kafka_dbg ( rk , GENERIC , " " , " " ) ; rd_kafka_q_enq ( rkb -> rkb_ops , rd_kafka_op_new ( RD_KAFKA_OP_TERMINATE ) ) ; rk -> rk_internal_rkb = NULL ; thrd = malloc ( sizeof ( * thrd ) ) ; * thrd = rkb -> rkb_thread ; rd_list_add ( & wait_thrds , thrd ) ; } mtx_unlock ( & rk -> rk_internal_rkb_lock ) ; if ( rkb ) rd_kafka_broker_destroy ( rkb ) ; rd_kafka_dbg ( rk , GENERIC , " " , " " , rd_list_cnt ( & wait_thrds ) ) ; RD_LIST_FOREACH ( thrd , & wait_thrds , i ) { int res ; if ( thrd_join ( * thrd , & res ) != thrd_success ) ; free ( thrd ) ; } rd_list_destroy ( & wait_thrds ) ; }
Main destructor for rd_kafka_t
 Locality: rdkafka main thread or application thread during rd_kafka_new()
Main destructor for rd_kafka_t Locality : rdkafka main thread or application thread during rd_kafka_new ()
/* Trigger any state-change waiters (which should check the
         * terminate flag whenever they wake up). */ /* Send op to trigger queue/io wake-up.
                 * The op itself is (likely) ignored by the receiver. */ /* Call on_destroy() interceptors */ /* Brokers pick up on rk_terminate automatically. */ /* List of (broker) threads to join to synchronize termination */ /* Decommission all topics */ /* Decommission brokers.
         * Broker thread holds a refcount and detects when broker refcounts
         * reaches 1 and then decommissions itself. */ /* Add broker's thread to wait_thrds list for later joining */ /* Send op to trigger queue/io wake-up.
                 * The op itself is (likely) ignored by the broker thread. */ /* Interrupt IO threads to speed up termination. */ /* Purge broker state change waiters */ /* Purge op-queue */ /* Loose our special reference to the internal broker. */ /* Send op to trigger queue wake-up. */ /* Join broker threads */