Fix refcnt preventing final destruction of a broker.

This issue happens when a broker is destroying and there are enqueued operations to add buffers to id.
If the operations are executed after purging the buffers their aren't destroyed and the refcnt for `rkbuf_rkb` prevents the broker and whole instance from being finally destroyed

Author: emasab

src/rdkafka_broker.c

@@ -4605,8 +4605,10 @@ static int rd_kafka_broker_thread_main(void *arg) {
                                 break;
                         }
 
-                        if (unlikely(rd_kafka_terminating(rkb->rkb_rk)))
+                        if (unlikely(rd_kafka_terminating(rkb->rkb_rk))) {
                                 rd_kafka_broker_serve(rkb, 1000);
+                                break;
+                        }
 
                         if (!rd_kafka_sasl_ready(rkb->rkb_rk)) {
                                 /* SASL provider not yet ready. */
@@ -4751,28 +4753,41 @@ static int rd_kafka_broker_thread_main(void *arg) {
                 }
         }
 
-        if (rkb->rkb_source != RD_KAFKA_INTERNAL) {
-                rd_kafka_wrlock(rkb->rkb_rk);
-                TAILQ_REMOVE(&rkb->rkb_rk->rk_brokers, rkb, rkb_link);
-                if (rkb->rkb_nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb))
-                        rd_list_remove(&rkb->rkb_rk->rk_broker_by_id, rkb);
-                (void)rd_atomic32_sub(&rkb->rkb_rk->rk_broker_cnt, 1);
-                rd_kafka_wrunlock(rkb->rkb_rk);
-        }
-
-        rd_kafka_broker_fail(rkb, LOG_DEBUG, RD_KAFKA_RESP_ERR__DESTROY,
-                             "Broker handle is terminating");
-
         /* Disable and drain ops queue.
          * Simply purging the ops queue risks leaving dangling references
          * for ops such as PARTITION_JOIN/PARTITION_LEAVE where the broker
          * reference is not maintained in the rko (but in rktp_next_leader).
-         * #1596 */
+         * #1596.
+         * Do this before failing the broker to make sure no buffers
+         * are enqueued after that. */
         rd_kafka_q_disable(rkb->rkb_ops);
         while (rd_kafka_broker_ops_serve(rkb, RD_POLL_NOWAIT))
                 ;
 
-        rd_kafka_broker_destroy(rkb);
+        rd_kafka_broker_fail(rkb, LOG_DEBUG, rd_kafka_broker_destroy_error(rk),
+                             "Broker handle is terminating");
+
+        rd_rkb_dbg(rkb, BROKER, "TERMINATE",
+                   "Handle terminates in state %s: "
+                   "%d refcnts (%p), %d toppar(s), "
+                   "%d active toppar(s), "
+                   "%d outbufs, %d waitresps, %d retrybufs",
+                   rd_kafka_broker_state_names[rkb->rkb_state],
+                   rd_refcnt_get(&rkb->rkb_refcnt), &rkb->rkb_refcnt,
+                   rkb->rkb_toppar_cnt, rkb->rkb_active_toppar_cnt,
+                   (int)rd_kafka_bufq_cnt(&rkb->rkb_outbufs),
+                   (int)rd_kafka_bufq_cnt(&rkb->rkb_waitresps),
+                   (int)rd_kafka_bufq_cnt(&rkb->rkb_retrybufs));
+
+        rd_dassert(rkb->rkb_state == RD_KAFKA_BROKER_STATE_DOWN);
+        if (rkb->rkb_source != RD_KAFKA_INTERNAL) {
+                rd_kafka_wrlock(rkb->rkb_rk);
+                TAILQ_REMOVE(&rkb->rkb_rk->rk_brokers, rkb, rkb_link);
+                if (rkb->rkb_nodeid != -1 && !RD_KAFKA_BROKER_IS_LOGICAL(rkb))
+                        rd_list_remove(&rkb->rkb_rk->rk_broker_by_id, rkb);
+                (void)rd_atomic32_sub(&rkb->rkb_rk->rk_broker_cnt, 1);
+                rd_kafka_wrunlock(rkb->rkb_rk);
+        }
 
 #if WITH_SSL
         /* Remove OpenSSL per-thread error state to avoid memory leaks */