Revert "[KIP-320] Validate assigned partitions before starting to consume from them (#4931)" (#5207)

This reverts commit 13a2bba.

Author: emasab

CHANGELOG.md

@@ -9,8 +9,6 @@ librdkafka v2.12.0 is a feature release:
 * Fix for KIP-1102 time based re-bootstrap condition (#5177).
 * Fix for discarding the member epoch in a consumer group heartbeat response when leaving with an inflight HB (#4672).
 * Fix for an error being raised after a commit due to an existing error in the topic partition (#4672).
-* Additional KIP-320 related validation: when assigning the offset is validated
-  if leader epoch is specified (#4931).
 * Fix double free of headers in `rd_kafka_produceva` method (@blindspotbounty, #4628).
 * Fix to ensure `rd_kafka_query_watermark_offsets` enforces the specified timeout and does not continue beyond timeout expiry (#5201).
 
@@ -40,18 +38,6 @@ librdkafka v2.12.0 is a feature release:
   Happening since 2.5.0. Since 2.10.1 unit tests are failing when run on
   big-endian architectures (#5183, @paravoid).
 
-### Consumer fixes
-
-* Issues: #5158.
-  Additional KIP-320 related validation: when assigning the offset is validated
-  if a leader epoch is specified. A committed offset could have been truncated
-  in case of unclean leader election and, when a different member starts
-  fetching from it on the new leader, it could get an offset out of range
-  and a subsequent offset reset. The assigned offset is now validated before
-  we start fetching from it and, in case it was truncated, fetching starts
-  from last available offset of given leader epoch.
-  Happens since 2.1.0 (#4931).
-
 ### Producer fixes
 
 * Issues: #4627.

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_cgrp.c

@@ -4430,9 +4430,6 @@ static void rd_kafka_cgrp_incr_unassign_done(rd_kafka_cgrp_t *rkcg) {
                              "unassign",
                              rkcg->rkcg_group_id->str);
                 rd_kafka_cgrp_unassign(rkcg);
-
-                /* Leave group, if desired. */
-                rd_kafka_cgrp_leave_maybe(rkcg);
                 return;
         }
 

src/rdkafka_mock.c

@@ -548,7 +548,7 @@ rd_kafka_mock_committed_offset_find(const rd_kafka_mock_partition_t *mpart,
 rd_kafka_mock_committed_offset_t *
 rd_kafka_mock_commit_offset(rd_kafka_mock_partition_t *mpart,
                             const rd_kafkap_str_t *group,
-                            rd_kafka_fetch_pos_t pos,
+                            int64_t offset,
                             const rd_kafkap_str_t *metadata) {
         rd_kafka_mock_committed_offset_t *coff;
 
@@ -571,13 +571,12 @@ rd_kafka_mock_commit_offset(rd_kafka_mock_partition_t *mpart,
 
         coff->metadata = rd_kafkap_str_copy(metadata);
 
-        coff->pos = pos;
+        coff->offset = offset;
 
         rd_kafka_dbg(mpart->topic->cluster->rk, MOCK, "MOCK",
-                     "Topic %s [%" PRId32
-                     "] committing offset %s"
+                     "Topic %s [%" PRId32 "] committing offset %" PRId64
                      " for group %.*s",
-                     mpart->topic->name, mpart->id, rd_kafka_fetch_pos2str(pos),
+                     mpart->topic->name, mpart->id, offset,
                      RD_KAFKAP_STR_PR(group));
 
         return coff;

src/rdkafka_mock_cgrp.c

@@ -252,21 +252,6 @@ rd_kafka_resp_err_t rd_kafka_mock_cgrp_classic_member_sync_set(
         return RD_KAFKA_RESP_ERR_NO_ERROR;
 }
 
-/**
- * @brief Member left the group
- *        either explicitly or due to a timeout.
- *        Trigger rebalance if there are still members left.
- *        If this was the last member, mark the group as empty.
- */
-void rd_kafka_mock_cgrp_classic_member_leave_rebalance_maybe(
-    rd_kafka_mock_cgrp_classic_t *mcgrp,
-    const char *reason) {
-        if (mcgrp->member_cnt > 0)
-                rd_kafka_mock_cgrp_classic_rebalance(mcgrp, reason);
-        else
-                rd_kafka_mock_cgrp_classic_set_state(
-                    mcgrp, RD_KAFKA_MOCK_CGRP_STATE_EMPTY, reason);
-}
 
 /**
  * @brief Member is explicitly leaving the group (through LeaveGroupRequest)
@@ -280,8 +265,7 @@ rd_kafka_resp_err_t rd_kafka_mock_cgrp_classic_member_leave(
 
         rd_kafka_mock_cgrp_classic_member_destroy(mcgrp, member);
 
-        rd_kafka_mock_cgrp_classic_member_leave_rebalance_maybe(
-            mcgrp, "explicit member leave");
+        rd_kafka_mock_cgrp_classic_rebalance(mcgrp, "explicit member leave");
 
         return RD_KAFKA_RESP_ERR_NO_ERROR;
 }
@@ -649,8 +633,7 @@ static void rd_kafka_mock_cgrp_classic_session_tmr_cb(rd_kafka_timers_t *rkts,
         }
 
         if (timeout_cnt)
-                rd_kafka_mock_cgrp_classic_member_leave_rebalance_maybe(
-                    mcgrp, "member timeout");
+                rd_kafka_mock_cgrp_classic_rebalance(mcgrp, "member timeout");
 }
 
 

src/rdkafka_mock_handlers.c

@@ -742,7 +742,6 @@ static int rd_kafka_mock_handle_ListOffsets(rd_kafka_mock_connection_t *mconn,
                         int32_t MaxNumOffsets;
                         rd_kafka_mock_partition_t *mpart = NULL;
                         rd_kafka_resp_err_t err          = all_err;
-                        int32_t LeaderEpoch              = -1;
 
                         rd_kafka_buf_read_i32(rkbuf, &Partition);
 
@@ -808,6 +807,7 @@ static int rd_kafka_mock_handle_ListOffsets(rd_kafka_mock_connection_t *mconn,
                         if (rkbuf->rkbuf_reqhdr.ApiVersion >= 4) {
                                 /* Response: LeaderEpoch */
                                 const rd_kafka_mock_msgset_t *mset = NULL;
+                                int32_t leader_epoch               = -1;
                                 rd_bool_t on_follower              = rd_false;
 
                                 if (mpart) {
@@ -818,12 +818,12 @@ static int rd_kafka_mock_handle_ListOffsets(rd_kafka_mock_connection_t *mconn,
                                         if (Offset >= 0 &&
                                             (mset = rd_kafka_mock_msgset_find(
                                                  mpart, Offset, on_follower))) {
-                                                LeaderEpoch =
+                                                leader_epoch =
                                                     mset->leader_epoch;
                                         }
                                 }
 
-                                rd_kafka_buf_write_i32(resp, LeaderEpoch);
+                                rd_kafka_buf_write_i32(resp, leader_epoch);
                         }
 
                         /* Response: Partition tags */
@@ -835,7 +835,7 @@ static int rd_kafka_mock_handle_ListOffsets(rd_kafka_mock_connection_t *mconn,
                                      "offset %" PRId64 " (leader epoch %" PRId32
                                      ") for %s: %s",
                                      RD_KAFKAP_STR_PR(&Topic), Partition,
-                                     Offset, LeaderEpoch,
+                                     Offset, mpart ? mpart->leader_epoch : -1,
                                      rd_kafka_offset2str(Timestamp),
                                      rd_kafka_err2str(err));
                 }
@@ -930,13 +930,12 @@ static int rd_kafka_mock_handle_OffsetFetch(rd_kafka_mock_connection_t *mconn,
                                     mpart, &GroupId);
 
                         /* Response: CommittedOffset */
-                        rd_kafka_buf_write_i64(resp,
-                                               coff ? coff->pos.offset : -1);
+                        rd_kafka_buf_write_i64(resp, coff ? coff->offset : -1);
 
                         if (rkbuf->rkbuf_reqhdr.ApiVersion >= 5) {
                                 /* Response: CommittedLeaderEpoch */
                                 rd_kafka_buf_write_i32(
-                                    resp, coff ? coff->pos.leader_epoch : -1);
+                                    resp, mpart ? mpart->leader_epoch : -1);
                         }
 
                         /* Response: Metadata */
@@ -953,11 +952,10 @@ static int rd_kafka_mock_handle_OffsetFetch(rd_kafka_mock_connection_t *mconn,
                                 rd_kafka_dbg(mcluster->rk, MOCK, "MOCK",
                                              "Topic %s [%" PRId32
                                              "] returning "
-                                             "committed offset %s"
+                                             "committed offset %" PRId64
                                              " for group %s",
                                              mtopic->name, mpart->id,
-                                             rd_kafka_fetch_pos2str(coff->pos),
-                                             coff->group);
+                                             coff->offset, coff->group);
                         else
                                 rd_kafka_dbg(mcluster->rk, MOCK, "MOCK",
                                              "Topic %.*s [%" PRId32
@@ -1111,7 +1109,6 @@ static int rd_kafka_mock_handle_OffsetCommit(rd_kafka_mock_connection_t *mconn,
                         rd_kafka_mock_partition_t *mpart = NULL;
                         rd_kafka_resp_err_t err          = all_err;
                         int64_t CommittedOffset;
-                        int32_t CommittedLeaderEpoch = -1;
                         rd_kafkap_str_t Metadata;
 
                         rd_kafka_buf_read_i32(rkbuf, &Partition);
@@ -1129,6 +1126,7 @@ static int rd_kafka_mock_handle_OffsetCommit(rd_kafka_mock_connection_t *mconn,
                         rd_kafka_buf_read_i64(rkbuf, &CommittedOffset);
 
                         if (rkbuf->rkbuf_reqhdr.ApiVersion >= 6) {
+                                int32_t CommittedLeaderEpoch;
                                 rd_kafka_buf_read_i32(rkbuf,
                                                       &CommittedLeaderEpoch);
 
@@ -1147,11 +1145,9 @@ static int rd_kafka_mock_handle_OffsetCommit(rd_kafka_mock_connection_t *mconn,
                         rd_kafka_buf_skip_tags(rkbuf);
 
                         if (!err)
-                                rd_kafka_mock_commit_offset(
-                                    mpart, &GroupId,
-                                    RD_KAFKA_FETCH_POS(CommittedOffset,
-                                                       CommittedLeaderEpoch),
-                                    &Metadata);
+                                rd_kafka_mock_commit_offset(mpart, &GroupId,
+                                                            CommittedOffset,
+                                                            &Metadata);
 
                         /* Response: ErrorCode */
                         rd_kafka_buf_write_i16(resp, err);

src/rdkafka_mock_int.h

@@ -296,7 +296,7 @@ typedef struct rd_kafka_mock_committed_offset_s {
         /**< mpart.committed_offsets */
         TAILQ_ENTRY(rd_kafka_mock_committed_offset_s) link;
         char *group;               /**< Allocated along with the struct */
-        rd_kafka_fetch_pos_t pos;  /**< Committed position */
+        int64_t offset;            /**< Committed offset */
         rd_kafkap_str_t *metadata; /**< Metadata, allocated separately */
 } rd_kafka_mock_committed_offset_t;
 
@@ -552,7 +552,7 @@ rd_kafka_mock_committed_offset_find(const rd_kafka_mock_partition_t *mpart,
 rd_kafka_mock_committed_offset_t *
 rd_kafka_mock_commit_offset(rd_kafka_mock_partition_t *mpart,
                             const rd_kafkap_str_t *group,
-                            rd_kafka_fetch_pos_t pos,
+                            int64_t offset,
                             const rd_kafkap_str_t *metadata);
 
 const rd_kafka_mock_msgset_t *

src/rdkafka_offset.c

@@ -1056,8 +1056,6 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
                             end_offset, end_offset_leader_epoch);
 
                 } else {
-                        rd_kafka_toppar_set_fetch_state(
-                            rktp, RD_KAFKA_TOPPAR_FETCH_VALIDATE_SEEK);
                         rd_kafka_toppar_unlock(rktp);
 
                         /* Seek to the updated end offset */

src/rdkafka_partition.c

@@ -38,10 +38,10 @@
 
 #include "rdunittest.h"
 
-const char *rd_kafka_fetch_states[] = {"none",          "stopping",
-                                       "stopped",       "offset-query",
-                                       "offset-wait",   "validate-epoch-wait",
-                                       "validate-seek", "active"};
+const char *rd_kafka_fetch_states[] = {"none",        "stopping",
+                                       "stopped",     "offset-query",
+                                       "offset-wait", "validate-epoch-wait",
+                                       "active"};
 
 
 static rd_kafka_op_res_t rd_kafka_toppar_op_serve(rd_kafka_t *rk,
@@ -2272,14 +2272,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 +2289,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

@@ -299,7 +299,6 @@ struct rd_kafka_toppar_s {                           /* rd_kafka_toppar_t */
                 RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY,
                 RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT,
                 RD_KAFKA_TOPPAR_FETCH_VALIDATE_EPOCH_WAIT,
-                RD_KAFKA_TOPPAR_FETCH_VALIDATE_SEEK,
                 RD_KAFKA_TOPPAR_FETCH_ACTIVE,
         } rktp_fetch_state; /* Broker thread's state */
 
@@ -620,8 +619,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,