Second version starting offset validation when partition already joined the consumer group

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

@@ -489,54 +489,29 @@ static int rd_kafka_assignment_serve_pending(rd_kafka_t *rk) {
                          *
                          * Start fetcher for partition and forward partition's
                          * fetchq to consumer group's queue. */
-                        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_dbg(rk, CGRP, "SRVPEND",
+                                     "Starting pending assigned partition "
+                                     "%s [%" PRId32 "] at %s",
+                                     rktpar->topic, rktpar->partition,
+                                     rd_kafka_fetch_pos2str(
+                                         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) {
-                                rd_kafka_dbg(
-                                    rk, CGRP, "SRVPEND",
-                                    "Validating assigned partition offset "
-                                    "%s [%" PRId32 "] at %s",
-                                    rktpar->topic, rktpar->partition,
-                                    rd_kafka_fetch_pos2str(pos));
-
-                                rd_kafka_toppar_forward_internal(
-                                    rktp, rk->rk_consumer.q);
-                                rd_kafka_toppar_lock(rktp);
-                                rd_kafka_toppar_set_fetch_state(
-                                    rktp,
-                                    RD_KAFKA_TOPPAR_FETCH_VALIDATE_EPOCH_WAIT);
-                                rd_kafka_toppar_set_next_fetch_position(rktp,
-                                                                        pos);
-                                rd_kafka_toppar_set_offset_validation_position(
-                                    rktp, pos);
-                                rd_kafka_offset_validate(rktp, "offset fetch");
-                                rd_kafka_toppar_unlock(rktp);
-
-                        } else {
-                                rd_kafka_dbg(
-                                    rk, CGRP, "SRVPEND",
-                                    "Starting pending assigned partition "
-                                    "%s [%" PRId32 "] at %s",
-                                    rktpar->topic, rktpar->partition,
-                                    rd_kafka_fetch_pos2str(pos));
-
-                                /* Start the fetcher */
-                                rktp->rktp_started = rd_true;
-                                rk->rk_consumer.assignment.started_cnt++;
-
-                                rd_kafka_toppar_op_fetch_start(
-                                    rktp, pos, rk->rk_consumer.q,
-                                    RD_KAFKA_NO_REPLYQ);
-                        }
+                        /* Start the fetcher */
+                        rktp->rktp_started = rd_true;
+                        rk->rk_consumer.assignment.started_cnt++;
+
+                        rd_kafka_toppar_op_fetch_start(
+                            rktp,
+                            rd_kafka_topic_partition_get_fetch_pos(rktpar),
+                            rk->rk_consumer.q, RD_KAFKA_NO_REPLYQ);
 
 
                 } else if (can_query_offsets) {

src/rdkafka_offset.c

@@ -1056,6 +1056,8 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
                             end_offset, end_offset_leader_epoch);
 
                 } else {
+                        /* It ensures a validation isn't started again
+                         * until the seek completes. */
                         rd_kafka_toppar_set_fetch_state(
                             rktp, RD_KAFKA_TOPPAR_FETCH_VALIDATE_SEEK);
                         rd_kafka_toppar_unlock(rktp);

src/rdkafka_partition.c

@@ -1748,6 +1748,16 @@ static void rd_kafka_toppar_fetch_start(rd_kafka_toppar_t *rktp,
                                       "no previously committed offset "
                                       "available");
 
+        } else if (pos.leader_epoch >= -1) {
+                rd_kafka_toppar_set_next_fetch_position(rktp, pos);
+                rd_kafka_toppar_set_offset_validation_position(rktp, pos);
+                /* This is necessary as it only starts
+                 * validating from states ACTIVE or
+                 * VALIDATE_EPOCH_WAIT. */
+                rd_kafka_toppar_set_fetch_state(
+                    rktp, RD_KAFKA_TOPPAR_FETCH_VALIDATE_EPOCH_WAIT);
+                rd_kafka_offset_validate(rktp, "validate fetch start position");
+
         } else {
                 rd_kafka_toppar_set_next_fetch_position(rktp, pos);
 
@@ -2272,14 +2282,7 @@ static void rd_kafka_toppar_op(rd_kafka_toppar_t *rktp,
         rd_kafka_toppar_op0(rktp, rko, replyq);
 }
 
-void rd_kafka_toppar_forward_internal(rd_kafka_toppar_t *rktp,
-                                      rd_kafka_q_t *fwdq) {
-        rd_kafka_q_lock(rktp->rktp_fetchq);
-        if (fwdq && !(rktp->rktp_fetchq->rkq_flags & RD_KAFKA_Q_F_FWD_APP))
-                rd_kafka_q_fwd_set0(rktp->rktp_fetchq, fwdq, 0, /* no do_lock */
-                                    0 /* no fwd_app */);
-        rd_kafka_q_unlock(rktp->rktp_fetchq);
-}
+
 
 /**
  * Start consuming partition (async operation).
@@ -2296,7 +2299,11 @@ rd_kafka_resp_err_t rd_kafka_toppar_op_fetch_start(rd_kafka_toppar_t *rktp,
                                                    rd_kafka_replyq_t replyq) {
         int32_t version;
 
-        rd_kafka_toppar_forward_internal(rktp, fwdq);
+        rd_kafka_q_lock(rktp->rktp_fetchq);
+        if (fwdq && !(rktp->rktp_fetchq->rkq_flags & RD_KAFKA_Q_F_FWD_APP))
+                rd_kafka_q_fwd_set0(rktp->rktp_fetchq, fwdq, 0, /* no do_lock */
+                                    0 /* no fwd_app */);
+        rd_kafka_q_unlock(rktp->rktp_fetchq);
 
         /* Bump version barrier. */
         version = rd_kafka_toppar_version_new_barrier(rktp);

src/rdkafka_partition.h

@@ -620,8 +620,6 @@ void rd_kafka_toppar_next_offset_handle(rd_kafka_toppar_t *rktp,
 void rd_kafka_toppar_broker_delegate(rd_kafka_toppar_t *rktp,
                                      rd_kafka_broker_t *rkb);
 
-void rd_kafka_toppar_forward_internal(rd_kafka_toppar_t *rktp,
-                                      rd_kafka_q_t *fwdq);
 
 rd_kafka_resp_err_t rd_kafka_toppar_op_fetch_start(rd_kafka_toppar_t *rktp,
                                                    rd_kafka_fetch_pos_t pos,