Release the whole broker when thread creation fails

[BNCMK]

The broker is abandoned, not only its descriptors, when thread creation fails

Follows #4456, which reports two pipe descriptors leaking when thrd_create()
fails in rd_kafka_broker_add(). That report is correct and its fix closes the
descriptors. The same line abandons everything else the broker owns, and this is
a separate defect on the same path.

Applies on top of the #4456 fix. src/rdkafka_broker.c, verified against 2.15.0.

What else is abandoned

rd_kafka_broker_add() allocates the broker, its wake-up pipe, its ops queue,
its averaging structures, its name, its locks and condition variables, and takes
two references on it, all before calling thrd_create(). On failure the function
reaches:

rd_free(rkb);

rd_free() frees the outer struct and nothing else. Everything
rd_kafka_broker_destroy_final() would release is left behind:

abandoned	consequence
rkb_origname, rkb_ApiVersions	heap
rkb_ops queue	heap, normally via rd_kafka_q_destroy_owner
six rd_avg structures	heap
locks, condition variables, refcount	never destroyed
the two references taken above	refcount bypassed entirely

The descriptors #4456 reports are the visible part, because they accumulate
somewhere a person can look. The rest is invisible until a process that retries
broker creation runs long enough to notice its heap.

Why the obvious fix aborts

Calling the destructor twice, once per reference, is the natural repair and it
fails:

Assertion `thrd_is_current(rkb->rkb_thread)' failed.

rd_kafka_broker_destroy_final() opens with that assertion because in every
other case it is reached from the broker's own thread. Here that thread was
never created, so the handle is uninitialised and the assertion fires on
whatever it contains.

The fix

At this point the broker is not yet on rk_brokers, rd_kafka_sasl_broker_init
has not run, and nothing else holds a pointer to it, so the calling thread is the
only thread that owns it. Recording that before tearing down makes the assertion
true rather than bypassed:

rkb->rkb_thread = thrd_current();
rd_kafka_broker_destroy(rkb); /* broker thread's refcnt */
rd_kafka_broker_destroy(rkb); /* rk_broker's refcnt */

Both references drop, the count reaches zero, and the destructor releases
everything including the descriptors, so the explicit closes added for #4456 are
no longer needed and this patch removes them.

The alternative is to keep closing resources inline as each is noticed. That is
how the path arrived here: the next resource added to a broker would have to be
remembered in two places, and the second place is the one nobody looks at.

If the assignment to rkb_thread is unwelcome, the same result comes from
relaxing the assertion for a broker whose thread was never started. That is a
maintainer's call about which invariant to state, and either way the teardown
belongs in one function.

Measured

Same reproduction as #4456, which counts /proc/self/fd before and after.

build	rounds	descriptors before	after	leaked
unpatched	5	3	23	20
with this patch	8	3	3	0
with this patch, MALLOC_CHECK_=3	12	3	3	0, no abort

Twelve rounds under MALLOC_CHECK_=3 report no double free, and the normal path
with thread creation succeeding is unchanged.

The diff

--- a/src/rdkafka_broker.c
+++ b/src/rdkafka_broker.c
@@ -5438,8 +5438,7 @@
          * the broker thread until we've finalized the rkb. */
         rd_kafka_broker_lock(rkb);
         rd_kafka_broker_keep(rkb); /* broker thread's refcnt */
-        if (getenv("RDK_FORCE_THRD_FAIL") ||
-            thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) !=
+        if (thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) !=
             thrd_success) {
                 rd_kafka_broker_unlock(rkb);
 
@@ -5450,19 +5449,29 @@
                 rd_kafka_op_err(rk, RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE,
                                 "Unable to create broker thread");
 
-                /* The wake-up pipe was opened above and is closed by
-                 * rd_kafka_broker_destroy_final(), which this path does not
-                 * reach: rd_free() releases the struct holding the descriptor
-                 * numbers without closing them, so nothing can close them
-                 * afterwards. Close them here before the struct goes. */
-#ifndef _WIN32
-                if (rkb->rkb_wakeup_fd[0] != -1)
-                        rd_socket_close(rkb->rkb_wakeup_fd[0]);
-                if (rkb->rkb_wakeup_fd[1] != -1)
-                        rd_socket_close(rkb->rkb_wakeup_fd[1]);
-#endif
+                /* Unwind through the destructor rather than by hand.
+                 *
+                 * rd_free() releases the struct without running
+                 * rd_kafka_broker_destroy_final(), so the wake-up pipe opened
+                 * above is never closed and the two descriptors stay open for
+                 * the life of the process with nothing left holding their
+                 * numbers. It also bypasses the refcount, leaving the two
+                 * references taken here unaccounted for.
+                 *
+                 * At this point the broker is not yet on rk_brokers and nothing
+                 * else can reach it, so dropping both references takes the count
+                 * to zero and runs the destructor, which closes the pipe, tears
+                 * down the locks and queues, and frees the struct. */
+                /* The destructor asserts it runs on the broker's own thread,
+                 * which is how it is reached in every other case. That thread
+                 * was never created, and no other thread can reach this broker
+                 * because it is not yet on rk_brokers, so this thread is the one
+                 * that owns it. Recording that makes the assertion true rather
+                 * than bypassed. */
+                rkb->rkb_thread = thrd_current();
 
-                rd_free(rkb);
+                rd_kafka_broker_destroy(rkb); /* broker thread's refcnt */
+                rd_kafka_broker_destroy(rkb); /* rk_broker's refcnt */
 
 #ifndef _WIN32
                 /* Restore sigmask of caller */

Tooling

The descriptor half was confirmed at the syscall boundary with
ExecVis, which records what every process on
a machine did without the program being instrumented, recompiled or aware. The
heap half was found by reading what the destructor releases and comparing it with
what the failure path does, which is the kind of comparison a leak checker only
reaches once the path has actually run, and this one runs only when thread
creation fails.

[confluent-cla-assistant]

Please sign the Contributor License Agreement here before this PR can be approved.
❌ BNCMK
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}