[KIP-714] Fix idle ratio calculation for non forwarded queues (confluentinc#5017)

* Fix for devel assert "rkq->rkq_ts_last_poll_end >= rkq->rkq_ts_last_poll_start" in test 0056.

Partition queues can be not forwarded to the main poll queue so these timestamps must be per-queue instead of per-instance. We avoid checking if app polled should be called in case of internal poll calls where we're sure it's not a consume call, we also avoid checking it if "app polled" was already called by a dedicated consume function.

Author: emasab

CHANGELOG.md

@@ -1,3 +1,23 @@
+# librdkafka v2.11.0
+
+librdkafka v2.11.0 is a feature release:
+
+* Fix for poll ratio calculation in case the queues are forwarded (#5017).
+
+
+## Fixes
+
+### Telemetry fixes
+
+* Issues: #5109
+  Fix for poll ratio calculation in case the queues are forwarded.
+  Poll ratio is now calculated per-queue instead of per-instance and
+  it allows to avoid calculation problems linked to using the same
+  field.
+  Happens since 2.6.0 (#5017).
+
+
+
 # librdkafka v2.10.1
 
 librdkafka v2.10.1 is a maintenance release:

src/rdkafka.c

@@ -3218,12 +3218,12 @@ static rd_kafka_op_res_t rd_kafka_consume_callback0(
         struct consume_ctx ctx = {.consume_cb = consume_cb, .opaque = opaque};
         rd_kafka_op_res_t res;
 
-        rd_kafka_app_poll_start(rkq->rkq_rk, 0, timeout_ms);
+        rd_kafka_app_poll_start(rkq->rkq_rk, rkq, 0, timeout_ms);
 
         res = rd_kafka_q_serve(rkq, timeout_ms, max_cnt, RD_KAFKA_Q_CB_RETURN,
                                rd_kafka_consume_cb, &ctx);
 
-        rd_kafka_app_polled(rkq->rkq_rk);
+        rd_kafka_app_polled(rkq->rkq_rk, rkq);
 
         return res;
 }
@@ -3289,7 +3289,7 @@ rd_kafka_consume0(rd_kafka_t *rk, rd_kafka_q_t *rkq, int timeout_ms) {
         rd_ts_t now                   = rd_clock();
         rd_ts_t abs_timeout           = rd_timeout_init0(now, timeout_ms);
 
-        rd_kafka_app_poll_start(rk, now, timeout_ms);
+        rd_kafka_app_poll_start(rk, rkq, now, timeout_ms);
 
         rd_kafka_yield_thread = 0;
         while ((
@@ -3306,7 +3306,7 @@ rd_kafka_consume0(rd_kafka_t *rk, rd_kafka_q_t *rkq, int timeout_ms) {
                         /* Callback called rd_kafka_yield(), we must
                          * stop dispatching the queue and return. */
                         rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__INTR, EINTR);
-                        rd_kafka_app_polled(rk);
+                        rd_kafka_app_polled(rk, rkq);
                         return NULL;
                 }
 
@@ -3318,7 +3318,7 @@ rd_kafka_consume0(rd_kafka_t *rk, rd_kafka_q_t *rkq, int timeout_ms) {
                 /* Timeout reached with no op returned. */
                 rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__TIMED_OUT,
                                         ETIMEDOUT);
-                rd_kafka_app_polled(rk);
+                rd_kafka_app_polled(rk, rkq);
                 return NULL;
         }
 
@@ -3333,7 +3333,7 @@ rd_kafka_consume0(rd_kafka_t *rk, rd_kafka_q_t *rkq, int timeout_ms) {
 
         rd_kafka_set_last_error(0, 0);
 
-        rd_kafka_app_polled(rk);
+        rd_kafka_app_polled(rk, rkq);
 
         return rkmessage;
 }
@@ -3980,10 +3980,10 @@ rd_kafka_op_res_t rd_kafka_poll_cb(rd_kafka_t *rk,
                     cb_type == RD_KAFKA_Q_CB_FORCE_RETURN)
                         return RD_KAFKA_OP_RES_PASS; /* Dont handle here */
                 else {
-                        rk->rk_ts_last_poll_end = rd_clock();
-                        struct consume_ctx ctx  = {.consume_cb =
-                                                       rk->rk_conf.consume_cb,
-                                                   .opaque = rk->rk_conf.opaque};
+                        rkq->rkq_ts_last_poll_end = rd_clock();
+                        struct consume_ctx ctx    = {.consume_cb =
+                                                         rk->rk_conf.consume_cb,
+                                                     .opaque = rk->rk_conf.opaque};
 
                         return rd_kafka_consume_cb(rk, rkq, rko, cb_type, &ctx);
                 }
@@ -4213,17 +4213,19 @@ rd_kafka_op_res_t rd_kafka_poll_cb(rd_kafka_t *rk,
 int rd_kafka_poll(rd_kafka_t *rk, int timeout_ms) {
         int r;
 
-        r = rd_kafka_q_serve(rk->rk_rep, timeout_ms, 0, RD_KAFKA_Q_CB_CALLBACK,
-                             rd_kafka_poll_cb, NULL);
+        r = rd_kafka_q_serve_maybe_consume(rk->rk_rep, timeout_ms, 0,
+                                           RD_KAFKA_Q_CB_CALLBACK,
+                                           rd_kafka_poll_cb, NULL);
         return r;
 }
 
 
 rd_kafka_event_t *rd_kafka_queue_poll(rd_kafka_queue_t *rkqu, int timeout_ms) {
         rd_kafka_op_t *rko;
 
-        rko = rd_kafka_q_pop_serve(rkqu->rkqu_q, rd_timeout_us(timeout_ms), 0,
-                                   RD_KAFKA_Q_CB_EVENT, rd_kafka_poll_cb, NULL);
+        rko = rd_kafka_q_pop_serve_maybe_consume(
+            rkqu->rkqu_q, rd_timeout_us(timeout_ms), 0, RD_KAFKA_Q_CB_EVENT,
+            rd_kafka_poll_cb, NULL);
 
 
         if (!rko)
@@ -4235,8 +4237,9 @@ rd_kafka_event_t *rd_kafka_queue_poll(rd_kafka_queue_t *rkqu, int timeout_ms) {
 int rd_kafka_queue_poll_callback(rd_kafka_queue_t *rkqu, int timeout_ms) {
         int r;
 
-        r = rd_kafka_q_serve(rkqu->rkqu_q, timeout_ms, 0,
-                             RD_KAFKA_Q_CB_CALLBACK, rd_kafka_poll_cb, NULL);
+        r = rd_kafka_q_serve_maybe_consume(rkqu->rkqu_q, timeout_ms, 0,
+                                           RD_KAFKA_Q_CB_CALLBACK,
+                                           rd_kafka_poll_cb, NULL);
         return r;
 }
 

src/rdkafka_int.h

@@ -392,16 +392,7 @@ struct rd_kafka_s {
                                         *   to avoid reaching
                                         * max.poll.interval.ms. during that time
                                         * frame. Only relevant for consumer. */
-        rd_ts_t rk_ts_last_poll_start; /**< Timestamp of last application
-                                        *   consumer_poll() call start
-                                        *   Only relevant for consumer.
-                                        *   Not an atomic as Kafka consumer
-                                        *   isn't thread safe. */
-        rd_ts_t rk_ts_last_poll_end;   /**< Timestamp of last application
-                                        *   consumer_poll() call end
-                                        *   Only relevant for consumer.
-                                        *   Not an atomic as Kafka consumer
-                                        *   isn't thread safe. */
+
         /* First fatal error. */
         struct {
                 rd_atomic32_t err; /**< rd_kafka_resp_err_t */
@@ -1179,29 +1170,31 @@ static RD_INLINE RD_UNUSED int rd_kafka_max_poll_exceeded(rd_kafka_t *rk) {
  * @locality any
  * @locks none
  */
-static RD_INLINE RD_UNUSED void
-rd_kafka_app_poll_start(rd_kafka_t *rk, rd_ts_t now, rd_bool_t is_blocking) {
+static RD_INLINE RD_UNUSED void rd_kafka_app_poll_start(rd_kafka_t *rk,
+                                                        rd_kafka_q_t *rkq,
+                                                        rd_ts_t now,
+                                                        rd_bool_t is_blocking) {
         if (rk->rk_type != RD_KAFKA_CONSUMER)
                 return;
 
         if (!now)
                 now = rd_clock();
         if (is_blocking)
                 rd_atomic64_set(&rk->rk_ts_last_poll, INT64_MAX);
-        if (rk->rk_ts_last_poll_end) {
+        if (rkq->rkq_ts_last_poll_end) {
                 int64_t poll_idle_ratio = 0;
-                rd_ts_t poll_interval   = now - rk->rk_ts_last_poll_start;
+                rd_ts_t poll_interval   = now - rkq->rkq_ts_last_poll_start;
                 if (poll_interval) {
-                        rd_ts_t idle_interval =
-                            rk->rk_ts_last_poll_end - rk->rk_ts_last_poll_start;
+                        rd_ts_t idle_interval = rkq->rkq_ts_last_poll_end -
+                                                rkq->rkq_ts_last_poll_start;
                         poll_idle_ratio =
                             idle_interval * 1000000 / poll_interval;
                 }
                 rd_avg_add(
                     &rk->rk_telemetry.rd_avg_current.rk_avg_poll_idle_ratio,
                     poll_idle_ratio);
-                rk->rk_ts_last_poll_start = now;
-                rk->rk_ts_last_poll_end   = 0;
+                rkq->rkq_ts_last_poll_start = now;
+                rkq->rkq_ts_last_poll_end   = 0;
         }
 }
 
@@ -1213,7 +1206,8 @@ rd_kafka_app_poll_start(rd_kafka_t *rk, rd_ts_t now, rd_bool_t is_blocking) {
  * @locality any
  * @locks none
  */
-static RD_INLINE RD_UNUSED void rd_kafka_app_polled(rd_kafka_t *rk) {
+static RD_INLINE RD_UNUSED void rd_kafka_app_polled(rd_kafka_t *rk,
+                                                    rd_kafka_q_t *rkq) {
         if (rk->rk_type == RD_KAFKA_CONSUMER) {
                 rd_ts_t now = rd_clock();
                 rd_atomic64_set(&rk->rk_ts_last_poll, now);
@@ -1226,11 +1220,10 @@ static RD_INLINE RD_UNUSED void rd_kafka_app_polled(rd_kafka_t *rk) {
                             rk->rk_cgrp,
                             "app polled after poll interval exceeded");
                 }
-                if (!rk->rk_ts_last_poll_end)
-                        rk->rk_ts_last_poll_end = now;
-                /* Disabled until #5017 is merged
-                 * rd_dassert(rk->rk_ts_last_poll_end >=
-                 *            rk->rk_ts_last_poll_start); */
+                if (!rkq->rkq_ts_last_poll_end)
+                        rkq->rkq_ts_last_poll_end = now;
+                rd_dassert(rkq->rkq_ts_last_poll_end >=
+                           rkq->rkq_ts_last_poll_start);
         }
 }