kafka: bump result of leader_epoch_last_offset to start offset

This could previously return a value that is below the start override.
Since the semantics are to return the next-highest term if the input term
falls below the log start, this patch does just that by bumping to the
effective start offset.

src/v/kafka/server/replicated_partition.cc

@@ -318,6 +318,23 @@ raft::replicate_stages replicated_partition::replicate(
 
 ss::future<std::optional<model::offset>>
 replicated_partition::get_leader_epoch_last_offset(
+  kafka::leader_epoch epoch) const {
+    auto offset_unbounded = co_await get_leader_epoch_last_offset_unbounded(
+      epoch);
+    if (!offset_unbounded.has_value()) {
+        co_return std::nullopt;
+    }
+    if (!_partition->kafka_start_offset_override().has_value()) {
+        co_return offset_unbounded;
+    }
+    // If the requested term falls below our earliest consumable segment as
+    // bounded by a start override, return the offset of the next-highest term
+    // (the new start offset).
+    co_return std::max(offset_unbounded.value(), start_offset());
+}
+
+ss::future<std::optional<model::offset>>
+replicated_partition::get_leader_epoch_last_offset_unbounded(
   kafka::leader_epoch epoch) const {
     const model::term_id term(epoch);
     const auto first_local_offset = _partition->raft_start_offset();

src/v/kafka/server/replicated_partition.h

@@ -209,6 +209,11 @@ class replicated_partition final : public kafka::partition_proxy::impl {
         return local_kafka_start_offset;
     }
 
+    // Returns the highest offset in the given term, without considering
+    // overrides of the starting offset.
+    ss::future<std::optional<model::offset>>
+      get_leader_epoch_last_offset_unbounded(kafka::leader_epoch) const;
+
     ss::future<std::vector<cluster::rm_stm::tx_range>>
       aborted_transactions_local(
         cloud_storage::offset_range,