Second version adding back the partition to the pending list while setting fetch_pos_validated to `rd_true'

Author: emasab

src/rdkafka.c

@@ -2984,6 +2984,10 @@ static RD_UNUSED int rd_kafka_consume_start0(rd_kafka_topic_t *rkt,
                 return -1;
         }
 
+        /*
+         * We cannot validate this offset as the legacy API doesn't
+         * provide the leader epoch.
+         */
         rd_kafka_toppar_op_fetch_start(rktp, RD_KAFKA_FETCH_POS(offset, -1),
                                        rkq, RD_KAFKA_NO_REPLYQ);
 

src/rdkafka_assignment.c

@@ -492,15 +492,10 @@ static int rd_kafka_assignment_serve_pending(rd_kafka_t *rk) {
                         rd_kafka_fetch_pos_t pos =
                             rd_kafka_topic_partition_get_fetch_pos(rktpar);
 
-                        /* Reset the (lib) pause flag which may have
-                         * been set by the cgrp when scheduling the
-                         * rebalance callback. */
-                        rd_kafka_toppar_op_pause_resume(
-                            rktp, rd_false /*resume*/,
-                            RD_KAFKA_TOPPAR_F_LIB_PAUSE, RD_KAFKA_NO_REPLYQ);
-
                         if (!RD_KAFKA_OFFSET_IS_LOGICAL(rktpar->offset) &&
-                            pos.leader_epoch != -1) {
+                            pos.leader_epoch != -1 &&
+                            !rd_kafka_topic_partition_get_fetch_pos_validated(
+                                rktpar)) {
                                 rd_kafka_dbg(
                                     rk, CGRP, "SRVPEND",
                                     "Validating assigned partition offset "
@@ -529,6 +524,14 @@ static int rd_kafka_assignment_serve_pending(rd_kafka_t *rk) {
                                     rktpar->topic, rktpar->partition,
                                     rd_kafka_fetch_pos2str(pos));
 
+                                /* Reset the (lib) pause flag which may have
+                                 * been set by the cgrp when scheduling the
+                                 * rebalance callback. */
+                                rd_kafka_toppar_op_pause_resume(
+                                    rktp, rd_false /*resume*/,
+                                    RD_KAFKA_TOPPAR_F_LIB_PAUSE,
+                                    RD_KAFKA_NO_REPLYQ);
+
                                 /* Start the fetcher */
                                 rktp->rktp_started = rd_true;
                                 rk->rk_consumer.assignment.started_cnt++;

src/rdkafka_offset.c

@@ -940,6 +940,7 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
         rd_kafka_toppar_lock(rktp);
         rktp->rktp_flags &= ~RD_KAFKA_TOPPAR_F_VALIDATING;
         rd_kafka_toppar_unlock(rktp);
+        rd_bool_t assignment_serve = rd_false;
 
         if (err == RD_KAFKA_RESP_ERR__DESTROY) {
                 rd_kafka_toppar_destroy(rktp); /* Drop refcnt */
@@ -1084,12 +1085,28 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
                            rktp->rktp_partition, end_offset,
                            end_offset_leader_epoch);
 
-                rd_kafka_toppar_set_fetch_state(rktp,
-                                                RD_KAFKA_TOPPAR_FETCH_ACTIVE);
+                if (rd_kafka_is_simple_consumer(rk) || rktp->rktp_started) {
+                        /* Already started, just set state to active */
+                        rd_kafka_toppar_set_fetch_state(
+                            rktp, RD_KAFKA_TOPPAR_FETCH_ACTIVE);
+                } else {
+                        rd_kafka_topic_partition_t *rktpar_pending =
+                            rd_kafka_topic_partition_list_add_copy(
+                                rk->rk_consumer.assignment.pending, rktpar);
+                        rd_kafka_topic_partition_set_from_fetch_pos(
+                            rktpar_pending, rktp->rktp_offset_validation_pos);
+                        rd_kafka_topic_partition_set_fetch_pos_validated(
+                            rktpar_pending, rd_true);
+                        /* Serve the assignment after releasing the lock
+                         * to respect lock order. */
+                        assignment_serve = rd_true;
+                }
         }
 
 done:
         rd_kafka_toppar_unlock(rktp);
+        if (assignment_serve)
+                rd_kafka_assignment_serve(rk);
 
         if (parts)
                 rd_kafka_topic_partition_list_destroy(parts);

src/rdkafka_partition.c

@@ -2864,6 +2864,44 @@ void rd_kafka_topic_partition_set_current_leader_epoch(
         parpriv->current_leader_epoch = current_leader_epoch;
 }
 
+/**
+ * @brief Marks the fetch position (offset + leader epoch) as validated or not.
+ * @param rktpar Partition object.
+ * @param fetch_pos_validated Whether the fetch position has been validated.
+ *
+ * @remark See KIP-320 for more information.
+ */
+void rd_kafka_topic_partition_set_fetch_pos_validated(
+    rd_kafka_topic_partition_t *rktpar,
+    rd_bool_t fetch_pos_validated) {
+        rd_kafka_topic_partition_private_t *parpriv;
+
+        /* Avoid allocating private_t if clearing the epoch */
+        if (!fetch_pos_validated || !rktpar->_private)
+                return;
+
+        parpriv = rd_kafka_topic_partition_get_private(rktpar);
+        parpriv->fetch_pos_validated = fetch_pos_validated;
+}
+
+/**
+ * @brief Gets the validation status of the fetch position (offset + leader
+ * epoch).
+ * @param rktpar Partition object.
+ * @return Whether the fetch position has been validated.
+ *
+ * @remark See KIP-320 for more information.
+ */
+rd_bool_t rd_kafka_topic_partition_get_fetch_pos_validated(
+    rd_kafka_topic_partition_t *rktpar) {
+        const rd_kafka_topic_partition_private_t *parpriv;
+
+        if (!(parpriv = rktpar->_private))
+                return rd_false;
+
+        return parpriv->fetch_pos_validated;
+}
+
 /**
  * @brief Set offset and leader epoch from a fetchpos.
  */

src/rdkafka_partition.h

@@ -499,6 +499,13 @@ typedef struct rd_kafka_topic_partition_private_s {
         int32_t current_leader_epoch;
         /** Leader epoch if known, else -1. */
         int32_t leader_epoch;
+        /**
+         * Is fetch position (offset + leader_epoch) validated?
+         * This field isn't copied automatically
+         * when copying or updating the struct to avoid unintended
+         * skipped validations.
+         */
+        rd_bool_t fetch_pos_validated;
         /** Topic id. */
         rd_kafka_Uuid_t topic_id;
 } rd_kafka_topic_partition_private_t;
@@ -848,30 +855,21 @@ rd_kafka_topic_partition_get_private(rd_kafka_topic_partition_t *rktpar) {
 }
 
 
-/**
- * @returns the partition leader current epoch, if relevant and known,
- *          else -1.
- *
- * @param rktpar Partition object.
- *
- * @remark See KIP-320 for more information.
- */
 int32_t rd_kafka_topic_partition_get_current_leader_epoch(
     const rd_kafka_topic_partition_t *rktpar);
 
 
-/**
- * @brief Sets the partition leader current epoch (use -1 to clear).
- *
- * @param rktpar Partition object.
- * @param leader_epoch Partition leader current epoch, use -1 to reset.
- *
- * @remark See KIP-320 for more information.
- */
 void rd_kafka_topic_partition_set_current_leader_epoch(
     rd_kafka_topic_partition_t *rktpar,
     int32_t leader_epoch);
 
+void rd_kafka_topic_partition_set_fetch_pos_validated(
+    rd_kafka_topic_partition_t *rktpar,
+    rd_bool_t fetch_pos_validated);
+
+rd_bool_t rd_kafka_topic_partition_get_fetch_pos_validated(
+    rd_kafka_topic_partition_t *rktpar);
+
 /**
  * @returns the partition's rktp if set (no refcnt increase), else NULL.
  */