Fix data race when a buffer queue is being initialized instead of being reset (confluentinc#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 confluentinc#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);
 }
 
 /**