Fix for #4427

confluentinc · Sep 26, 2023 · 232a00e · 232a00e
1 parent ae56961
commit 232a00e
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 19 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,6 +8,18 @@ librdkafka v2.2.1 is a maintenance release:
    are partition leader changes and a stale leader epoch is received (#4429).
  * Fix a segmentation fault when closing a consumer using the
    cooperative-sticky assignor before the first assignment (#4381).
+ * Fix to ensure permanent errors during offset validation continue being retried and
+   don't cause an offset reset (#).
+
+
+## Fixes
+
+### Consumer fixes
+
+ * During offset validation a permanent error like host resolution failure
+   would cause an offset reset.
+   This isn't what's expected or what the Java implementation does.
+   Solved by retrying even in case of permanent errors.
 
 
 

diff --git a/src/rdkafka_offset.c b/src/rdkafka_offset.c
@@ -991,25 +991,15 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
                         rd_kafka_topic_leader_query0(rk, rktp->rktp_rkt, 1,
                                                      rd_true /* force */);
 
-                if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
-                        /* No need for refcnt on rktp for timer opaque
-                         * since the timer resides on the rktp and will be
-                         * stopped on toppar remove. */
-                        rd_kafka_timer_start_oneshot(
-                            &rk->rk_timers, &rktp->rktp_validate_tmr, rd_false,
-                            500 * 1000 /* 500ms */,
-                            rd_kafka_offset_validate_tmr_cb, rktp);
-
-                } else if (actions & RD_KAFKA_ERR_ACTION_PERMANENT) {
-                        /* Permanent error */
-                        rd_kafka_offset_reset(
-                            rktp, rd_kafka_broker_id(rkb),
-                            RD_KAFKA_FETCH_POS(RD_KAFKA_OFFSET_INVALID,
-                                               rktp->rktp_leader_epoch),
-                            RD_KAFKA_RESP_ERR__LOG_TRUNCATION,
-                            "Unable to validate offset and epoch: %s",
-                            rd_kafka_err2str(err));
-                }
+                /* No need for refcnt on rktp for timer opaque
+                 * since the timer resides on the rktp and will be
+                 * stopped on toppar remove.
+                 * Retries the validation with a new call even in
+                 * case of permanent error. */
+                rd_kafka_timer_start_oneshot(
+                    &rk->rk_timers, &rktp->rktp_validate_tmr, rd_false,
+                    500 * 1000 /* 500ms */, rd_kafka_offset_validate_tmr_cb,
+                    rktp);
                 goto done;
         }