Remove next leader epoch

using a fetch state check instead.
Fix stale next fetch position update
when fetch hasn't started yet

src/rdkafka_offset.c

@@ -1013,10 +1013,8 @@ static void rd_kafka_toppar_handle_OffsetForLeaderEpoch(rd_kafka_t *rk,
         }
 
 
-        /* Validation succeeded, replace leader epoch */
-        rktp->rktp_leader_epoch = rktp->rktp_next_leader_epoch;
-        rktpar                  = &parts->elems[0];
-        end_offset              = rktpar->offset;
+        rktpar     = &parts->elems[0];
+        end_offset = rktpar->offset;
         end_offset_leader_epoch =
             rd_kafka_topic_partition_get_leader_epoch(rktpar);
 
@@ -1142,6 +1140,28 @@ void rd_kafka_offset_validate(rd_kafka_toppar_t *rktp, const char *fmt, ...) {
                 return;
         }
 
+
+        if (rktp->rktp_leader_id == -1 || !rktp->rktp_leader ||
+            rktp->rktp_leader->rkb_source == RD_KAFKA_INTERNAL) {
+                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, FETCH, "VALIDATE",
+                             "%.*s [%" PRId32
+                             "]: unable to perform offset "
+                             "validation: partition leader not available",
+                             RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
+                             rktp->rktp_partition);
+
+                rd_kafka_toppar_set_fetch_state(rktp,
+                                                RD_KAFKA_TOPPAR_FETCH_ACTIVE);
+                return;
+        }
+
+        /* Update next fetch position, that could be stale since last
+         * fetch start. Only if the offset is real. */
+        if (rktp->rktp_offsets.fetch_pos.offset > 0) {
+                rd_kafka_toppar_set_next_fetch_position(
+                    rktp, rktp->rktp_offsets.fetch_pos);
+        }
+
         /* If the fetch start position does not have an epoch set then
          * there is no point in doing validation.
          * This is the case for epoch-less seek()s or epoch-less
@@ -1155,18 +1175,6 @@ void rd_kafka_offset_validate(rd_kafka_toppar_t *rktp, const char *fmt, ...) {
                     RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                     rktp->rktp_partition,
                     rd_kafka_fetch_pos2str(rktp->rktp_next_fetch_start));
-                return;
-        }
-
-        if (rktp->rktp_leader_id == -1 || !rktp->rktp_leader ||
-            rktp->rktp_leader->rkb_source == RD_KAFKA_INTERNAL) {
-                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, FETCH, "VALIDATE",
-                             "%.*s [%" PRId32
-                             "]: unable to perform offset "
-                             "validation: partition leader not available",
-                             RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
-                             rktp->rktp_partition);
-
                 rd_kafka_toppar_set_fetch_state(rktp,
                                                 RD_KAFKA_TOPPAR_FETCH_ACTIVE);
                 return;
@@ -1175,17 +1183,14 @@ void rd_kafka_offset_validate(rd_kafka_toppar_t *rktp, const char *fmt, ...) {
         rd_kafka_toppar_set_fetch_state(
             rktp, RD_KAFKA_TOPPAR_FETCH_VALIDATE_EPOCH_WAIT);
 
-        rd_kafka_toppar_set_next_fetch_position(rktp,
-                                                rktp->rktp_offsets.fetch_pos);
-
         /* Construct and send OffsetForLeaderEpochRequest */
         parts  = rd_kafka_topic_partition_list_new(1);
         rktpar = rd_kafka_topic_partition_list_add(
             parts, rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition);
         rd_kafka_topic_partition_set_leader_epoch(
             rktpar, rktp->rktp_next_fetch_start.leader_epoch);
         rd_kafka_topic_partition_set_current_leader_epoch(
-            rktpar, rktp->rktp_next_leader_epoch);
+            rktpar, rktp->rktp_leader_epoch);
         rd_kafka_toppar_keep(rktp); /* for request opaque */
 
         rd_rkb_dbg(rktp->rktp_leader, FETCH, "VALIDATE",

src/rdkafka_partition.c

@@ -218,12 +218,11 @@ rd_kafka_toppar_t *rd_kafka_toppar_new0(rd_kafka_topic_t *rkt,
 
         rktp = rd_calloc(1, sizeof(*rktp));
 
-        rktp->rktp_partition         = partition;
-        rktp->rktp_rkt               = rkt;
-        rktp->rktp_leader_id         = -1;
-        rktp->rktp_broker_id         = -1;
-        rktp->rktp_leader_epoch      = -1;
-        rktp->rktp_next_leader_epoch = -1;
+        rktp->rktp_partition    = partition;
+        rktp->rktp_rkt          = rkt;
+        rktp->rktp_leader_id    = -1;
+        rktp->rktp_broker_id    = -1;
+        rktp->rktp_leader_epoch = -1;
         rd_interval_init(&rktp->rktp_lease_intvl);
         rd_interval_init(&rktp->rktp_new_lease_intvl);
         rd_interval_init(&rktp->rktp_new_lease_log_intvl);

src/rdkafka_partition.h

@@ -291,13 +291,8 @@ struct rd_kafka_toppar_s {                           /* rd_kafka_toppar_t */
 #define RD_KAFKA_TOPPAR_FETCH_IS_STARTED(fetch_state)                          \
         ((fetch_state) >= RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY)
 
-        int32_t rktp_leader_epoch;      /**< Last known partition leader epoch,
-                                         *   or -1. */
-        int32_t rktp_next_leader_epoch; /**< Next leader epoch,
-                                         *  to replace after offset
-                                         *  validation */
-
-
+        int32_t rktp_leader_epoch; /**< Last known partition leader epoch,
+                                    *   or -1. */
 
         int32_t rktp_fetch_msg_max_bytes; /* Max number of bytes to
                                            * fetch.

src/rdkafka_topic.c

@@ -675,16 +675,18 @@ static int rd_kafka_toppar_leader_update(rd_kafka_topic_t *rkt,
                 return 0;
         }
 
-        if (leader_epoch > rktp->rktp_leader_epoch) {
+        if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_VALIDATE_EPOCH_WAIT)
+                need_epoch_validation = rd_true;
+        else if (leader_epoch > rktp->rktp_leader_epoch) {
                 rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BROKER",
                              "%s [%" PRId32 "]: leader %" PRId32
                              " epoch %" PRId32 " -> leader %" PRId32
                              " epoch %" PRId32,
                              rktp->rktp_rkt->rkt_topic->str,
                              rktp->rktp_partition, rktp->rktp_leader_id,
                              rktp->rktp_leader_epoch, leader_id, leader_epoch);
-                rktp->rktp_next_leader_epoch = leader_epoch;
-                need_epoch_validation        = rd_true;
+                rktp->rktp_leader_epoch = leader_epoch;
+                need_epoch_validation   = rd_true;
         }
 
         fetching_from_follower =