Fix data race when a buffer queue is being initialized instead of being reset (#4718)

A data race happened when emptying buffers of a failing broker, in its thread,
with the statistics callback in main thread gathering the buffer counts.
Solved by resetting the atomic counters instead of initializing them.
Happening since 1.x

Closes #4522

Author: emasab

CHANGELOG.md

@@ -4,10 +4,20 @@ librdkafka v2.11.0 is a feature release:
 
 * [KIP-1102](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1102%3A+Enable+clients+to+rebootstrap+based+on+timeout+or+error+code) Enable clients to rebootstrap based on timeout or error code (#4981).
 * Fix for poll ratio calculation in case the queues are forwarded (#5017).
+* Fix data race when buffer queues are being reset instead of being
+  initialized (#4718).
 
 
 ## Fixes
 
+### General fixes
+
+* Issues: #4522.
+  A data race happened when emptying buffers of a failing broker, in its thread,
+  with the statistics callback in main thread gathering the buffer counts.
+  Solved by resetting the atomic counters instead of initializing them.
+  Happening since 1.x (#4718).
+
 ### Telemetry fixes
 
 * Issues: #5109

src/rdkafka_buf.c

@@ -241,6 +241,12 @@ void rd_kafka_bufq_init(rd_kafka_bufq_t *rkbufq) {
         rd_atomic32_init(&rkbufq->rkbq_msg_cnt, 0);
 }
 
+static void rd_kafka_bufq_reset(rd_kafka_bufq_t *rkbufq) {
+        TAILQ_INIT(&rkbufq->rkbq_bufs);
+        rd_atomic32_set(&rkbufq->rkbq_cnt, 0);
+        rd_atomic32_set(&rkbufq->rkbq_msg_cnt, 0);
+}
+
 /**
  * Concat all buffers from 'src' to tail of 'dst'
  */
@@ -249,7 +255,7 @@ void rd_kafka_bufq_concat(rd_kafka_bufq_t *dst, rd_kafka_bufq_t *src) {
         (void)rd_atomic32_add(&dst->rkbq_cnt, rd_atomic32_get(&src->rkbq_cnt));
         (void)rd_atomic32_add(&dst->rkbq_msg_cnt,
                               rd_atomic32_get(&src->rkbq_msg_cnt));
-        rd_kafka_bufq_init(src);
+        rd_kafka_bufq_reset(src);
 }
 
 /**