Close wakeup fds on thread failure

[BNCMK]

File descriptors leak when broker thread creation fails

Confirms #4456, identifies the mechanism, and includes a reproduction and a
one-hunk fix. src/rdkafka_broker.c, verified against 2.15.0.

What leaks

rd_kafka_broker_add() opens the broker's wake-up pipe, then starts the broker
thread:

• the pipe is created by rd_pipe_nonblocking(rkb->rkb_wakeup_fd)
• rkb_wakeup_fd[0] and [1] are closed in rd_kafka_broker_destroy_final()
• on thrd_create() failure the function calls rd_free(rkb) and returns NULL

rd_free() is a plain free. It does not run rd_kafka_broker_destroy_final(),
so neither descriptor is closed, and the struct holding their numbers is released
at the same moment. Nothing can close them afterwards either: they stay open for
the life of the process with no reference to them anywhere.

Every failed rd_kafka_broker_add() therefore leaks exactly two descriptors, and
each is a pipe, which matches the "unreleased FIFO pipes" wording in the report.

Reproduction

repro_4456.c counts /proc/self/fd before and after and reports how many are
pipes, so the figure comes from the kernel rather than from the library's own
bookkeeping.

Triggering the path needs thrd_create() to fail. Two approaches do not work and
are worth recording so nobody repeats them:

• RLIMIT_NPROC fails the client's own threads first, so the run never reaches
  rd_kafka_broker_add().
• Interposing pthread_create with LD_PRELOAD has no effect: librdkafka.so
  carries no dynamic reference to it, because glibc 2.34 and later resolve it
  internally.

The failure was forced with a temporary condition on the thrd_create call, which
is not part of the patch:

if (getenv("RDK_FORCE_THRD_FAIL") ||
    thrd_create(&rkb->rkb_thread, rd_kafka_broker_thread_main, rkb) != thrd_success) {

Results, same binary, same reproduction, 5 clients:

build	descriptors before	after	leaked
unpatched	3	23	20
patched	3	3	0

Two descriptors per broker, two brokers per client, five clients. With the patch
the count is unchanged after 8 rounds, and a normal run with thread creation
succeeding is also unchanged, so the working path is undisturbed.

Confirmed independently at the syscall boundary

Watching the reproduction from outside the process, with no knowledge of
librdkafka's internals:

repro_4456: pipe2 10, close 9, openat 15
  descriptors opened 35, closed 9, never closed 26

Ten pipe2 calls: five clients, two brokers each, one wake-up pipe per broker.
Twenty of those descriptors are never closed, which is the same figure the
/proc/self/fd count reports, reached from the other direction.

The diff

--- a/src/rdkafka_broker.c
+++ b/src/rdkafka_broker.c
@@ -5450,6 +5450,18 @@
                 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
+
                 rd_free(rkb);
 
 #ifndef _WIN32

Tooling

Found by reading the failure path, then confirmed at the syscall boundary with
ExecVis, which records what every process on a
machine did without the program being instrumented, recompiled or aware.

This is the kind of defect it exists to make visible: the leak only runs when
thread creation fails, so no test exercises it, no log records it, and once
rd_free runs the descriptors have no owner left to report them. Counting opens
against closes per process finds the imbalance regardless.

[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.}