[CORE-14911] Fix _schemas replay for a truncated log

src/v/cluster/errc.h

@@ -104,6 +104,7 @@ enum class errc : int16_t {
     invalid_target_node_id,
     topic_id_already_exists,
     feature_sanctioned,
+    offset_out_of_range,
 };
 
 fmt::iterator format_to(errc err, fmt::iterator);
@@ -298,6 +299,9 @@ struct errc_category final : public std::error_category {
             return "A topic with the given id already exists";
         case errc::feature_sanctioned:
             return "Unable to use requested feature - license is invalid";
+        case errc::offset_out_of_range:
+            return "Requested offset is outside of the available range of the "
+                   "partition";
             REDPANDA_BEGIN_IGNORE_DEPRECATIONS
             // unused in the codebase but still in the enum, include it here
             // since clang wants switches to be exhaustive

src/v/cluster/errors.cc

@@ -213,6 +213,8 @@ fmt::iterator format_to(errc err, fmt::iterator out) {
         return fmt::format_to(out, "cluster::errc::topic_id_already_exists");
     case errc::feature_sanctioned:
         return fmt::format_to(out, "cluster::errc::feature_sanctioned");
+    case errc::offset_out_of_range:
+        return fmt::format_to(out, "cluster::errc::offset_out_of_range");
         REDPANDA_BEGIN_IGNORE_DEPRECATIONS
     case errc::inconsistent_stm_update:
         return fmt::format_to(

src/v/kafka/data/rpc/serde.cc

@@ -81,9 +81,10 @@ fmt::iterator topic_partitions::format_to(fmt::iterator it) const {
 fmt::iterator partition_offsets::format_to(fmt::iterator it) const {
     return fmt::format_to(
       it,
-      "{{ high_watermark: {}, last_stable_offset: {} }}",
+      "{{ high_watermark: {}, last_stable_offset: {}, start_offset: {} }}",
       high_watermark,
-      last_stable_offset);
+      last_stable_offset,
+      start_offset);
 }
 
 fmt::iterator partition_offset_result::format_to(fmt::iterator it) const {

src/v/kafka/data/rpc/serde.h

@@ -121,11 +121,14 @@ struct topic_partitions
 
 struct partition_offsets
   : serde::
-      envelope<partition_offsets, serde::version<0>, serde::compat_version<0>> {
-    auto serde_fields() { return std::tie(high_watermark, last_stable_offset); }
+      envelope<partition_offsets, serde::version<1>, serde::compat_version<0>> {
+    auto serde_fields() {
+        return std::tie(high_watermark, last_stable_offset, start_offset);
+    }
 
     kafka::offset high_watermark;
     kafka::offset last_stable_offset;
+    std::optional<kafka::offset> start_offset;
 
     fmt::iterator format_to(fmt::iterator it) const;
 };

src/v/kafka/data/rpc/service.cc

@@ -20,6 +20,7 @@
 #include "model/record_batch_reader.h"
 #include "model/timeout_clock.h"
 #include "raft/errc.h"
+#include "storage/exceptions.h"
 
 #include <seastar/core/chunked_fifo.hh>
 #include <seastar/core/future.hh>
@@ -159,6 +160,7 @@ local_service::get_partition_offsets(
           return ssx::now<ret_t>(partition_offsets{
             .high_watermark = model::offset_cast(partition->high_watermark()),
             .last_stable_offset = model::offset_cast(lso_r.value()),
+            .start_offset = model::offset_cast(partition->start_offset()),
           });
       });
 }
@@ -212,6 +214,25 @@ local_service::consume(
               co_return cluster::errc::not_leader;
           }
 
+          auto deadline = model::timeout_clock::now() + timeout;
+
+          auto offset_ec = co_await partition->validate_fetch_offset(
+            kafka::offset_cast(start_offset),
+            /*is_follower=*/false,
+            deadline);
+          if (offset_ec == kafka::error_code::offset_out_of_range) {
+              co_return cluster::errc::offset_out_of_range;
+          }
+          if (offset_ec != kafka::error_code::none) {
+              vlog(
+                log.warn,
+                "Error validating fetch offset {} on partition {}: {}",
+                start_offset,
+                partition->ntp(),
+                offset_ec);
+              co_return cluster::errc::partition_operation_failed;
+          }
+
           // Create log reader config
           kafka::log_reader_config reader_cfg(
             start_offset,
@@ -223,23 +244,30 @@ local_service::consume(
             std::nullopt,  // client_address
             false);        // strict_max_bytes
 
-          auto deadline = model::timeout_clock::now() + timeout;
-
-          // Create reader
-          auto translating_reader = co_await partition->make_reader(reader_cfg);
-
-          // Consume batches from reader
+          // Create reader and consume batches from it
           try {
+              auto translating_reader = co_await partition->make_reader(
+                reader_cfg);
               co_return co_await model::consume_reader_to_chunked_vector(
                 std::move(translating_reader.reader), deadline);
           } catch (const ss::timed_out_error&) {
               co_return cluster::errc::timeout;
+          } catch (const translation_offset_out_of_range& e) {
+              vlog(
+                log.warn,
+                "Offset {} is outside the translation range of partition {}: "
+                "{}",
+                start_offset,
+                partition->ntp(),
+                e.what());
+              co_return cluster::errc::offset_out_of_range;
           } catch (...) {
+              auto eptr = std::current_exception();
               vlog(
                 log.warn,
                 "Error consuming from partition {}: {}",
                 partition->ntp(),
-                std::current_exception());
+                eptr);
               co_return cluster::errc::partition_operation_failed;
           }
       });

src/v/kafka/data/rpc/test/deps.h

@@ -78,13 +78,19 @@ struct produced_batch {
 class in_memory_proxy : public kafka::partition_proxy::impl {
 public:
     in_memory_proxy(
-      const model::ktp& ktp, ss::chunked_fifo<produced_batch>* produced_batches)
+      const model::ktp& ktp,
+      ss::chunked_fifo<produced_batch>* produced_batches,
+      model::offset start_offset = model::offset{0})
       : _ntp(ktp.to_ntp())
-      , _produced_batches(produced_batches) {}
+      , _produced_batches(produced_batches)
+      , _start_offset(start_offset) {}
     in_memory_proxy(
-      model::ntp ntp, ss::chunked_fifo<produced_batch>* produced_batches)
+      model::ntp ntp,
+      ss::chunked_fifo<produced_batch>* produced_batches,
+      model::offset start_offset = model::offset{0})
       : _ntp(std::move(ntp))
-      , _produced_batches(produced_batches) {}
+      , _produced_batches(produced_batches)
+      , _start_offset(start_offset) {}
 
     const model::ntp& ntp() const final { return _ntp; }
     ss::future<result<model::offset, kafka::error_code>>
@@ -94,9 +100,7 @@ class in_memory_proxy : public kafka::partition_proxy::impl {
     model::offset local_start_offset() const final {
         throw std::runtime_error("unimplemented");
     }
-    model::offset start_offset() const final {
-        throw std::runtime_error("unimplemented");
-    }
+    model::offset start_offset() const final { return _start_offset; }
     model::offset high_watermark() const final {
         return model::next_offset(latest_offset());
     }
@@ -151,8 +155,12 @@ class in_memory_proxy : public kafka::partition_proxy::impl {
         throw std::runtime_error("unimplemented");
     }
     ss::future<kafka::error_code> validate_fetch_offset(
-      model::offset, bool, model::timeout_clock::time_point) final {
-        throw std::runtime_error("unimplemented");
+      model::offset fetch_offset,
+      bool,
+      model::timeout_clock::time_point) final {
+        co_return fetch_offset < _start_offset
+          ? kafka::error_code::offset_out_of_range
+          : kafka::error_code::none;
     }
 
     ss::future<result<model::offset>> replicate(
@@ -215,6 +223,7 @@ class in_memory_proxy : public kafka::partition_proxy::impl {
 
     model::ntp _ntp;
     ss::chunked_fifo<produced_batch>* _produced_batches;
+    model::offset _start_offset;
 };
 
 class fake_partition_leader_cache : public partition_leader_cache {
@@ -428,6 +437,10 @@ class fake_partition_manager_proxy {
 
     void set_errors(int n) { _errors_to_inject = n; }
 
+    void set_start_offset(const model::ntp& ntp, model::offset o) {
+        _start_offsets.insert_or_assign(ntp, o);
+    }
+
     void set_shard_owner(const model::ntp& ntp, ss::shard_id shard_id) {
         _shard_locations.insert_or_assign(ntp, shard_id);
     }
@@ -464,7 +477,8 @@ class fake_partition_manager_proxy {
             co_await _stall_cv.wait([this] { return !_stalled; });
         }
         auto pp = kafka::partition_proxy(
-          std::make_unique<in_memory_proxy>(ntp, &_produced_batches));
+          std::make_unique<in_memory_proxy>(
+            ntp, &_produced_batches, start_offset(ntp)));
         co_return co_await fn(&pp);
     }
 
@@ -473,11 +487,18 @@ class fake_partition_manager_proxy {
     }
 
 private:
+    template<typename N>
+    model::offset start_offset(const N& ntp) const {
+        auto it = _start_offsets.find(ntp);
+        return it == _start_offsets.end() ? model::offset{0} : it->second;
+    }
+
     int _errors_to_inject = 0;
     bool _stalled{false};
     ss::condition_variable _stall_cv;
     ss::chunked_fifo<produced_batch> _produced_batches;
     model::ntp_map_type<ss::shard_id> _shard_locations;
+    model::ntp_map_type<model::offset> _start_offsets;
 };
 
 class fake_partition_manager : public partition_manager {
@@ -493,6 +514,10 @@ class fake_partition_manager : public partition_manager {
 
     void set_errors(int n) { _fake_proxy->set_errors(n); }
 
+    void set_start_offset(const model::ntp& ntp, model::offset o) {
+        _fake_proxy->set_start_offset(ntp, o);
+    }
+
     void set_shard_owner(const model::ntp& ntp, ss::shard_id shard_id) {
         _fake_proxy->set_shard_owner(ntp, shard_id);
     }

src/v/kafka/server/errors.h

@@ -52,6 +52,8 @@ constexpr error_code map_topic_error_code(cluster::errc code) {
         return error_code::not_coordinator;
     case cluster::errc::invalid_request:
         return error_code::invalid_request;
+    case cluster::errc::offset_out_of_range:
+        return error_code::offset_out_of_range;
     case cluster::errc::throttling_quota_exceeded:
         return error_code::throttling_quota_exceeded;
     case cluster::errc::update_in_progress:

src/v/pandaproxy/schema_registry/BUILD

@@ -105,6 +105,7 @@ redpanda_cc_library(
         "//src/v/kafka/client:exceptions",
         "//src/v/kafka/data/rpc",
         "//src/v/kafka/protocol:create_topics",
+        "//src/v/kafka/protocol:list_offset",
         "//src/v/kafka/server:topic_config_utils",
         "//src/v/model",
         "//src/v/pandaproxy:logger",
@@ -299,6 +300,7 @@ redpanda_cc_library(
     ],
     implementation_deps = [
         "//src/v/base",
+        "//src/v/kafka/protocol",
         "//src/v/pandaproxy:logger",
         "//src/v/storage:record_batch_builder",
     ],

src/v/pandaproxy/schema_registry/kafka_client_transport.cc

@@ -25,6 +25,7 @@
 #include "kafka/client/exceptions.h"
 #include "kafka/data/rpc/deps.h"
 #include "kafka/protocol/create_topics.h"
+#include "kafka/protocol/list_offset.h"
 #include "kafka/server/handlers/topics/types.h"
 #include "model/namespace.h"
 #include "pandaproxy/logger.h"
@@ -132,8 +133,26 @@ kafka_client_transport::produce(model::record_batch batch) {
 }
 
 ss::future<model::offset> kafka_client_transport::get_high_watermark() {
+    return list_offset(kafka::list_offsets_request::latest_timestamp);
+}
+
+ss::future<model::offset> kafka_client_transport::get_log_start() {
+    return list_offset(kafka::list_offsets_request::earliest_timestamp);
+}
+
+ss::future<model::offset>
+kafka_client_transport::list_offset(model::timestamp timestamp) {
+    kafka::list_offsets_request req;
+    req.data.topics.emplace_back(
+      kafka::list_offset_topic{
+        .name = model::schema_registry_internal_tp.topic,
+        .partitions = {kafka::list_offset_partition{
+          .partition_index = model::schema_registry_internal_tp.partition,
+          .timestamp = timestamp,
+          .max_num_offsets = 1,
+        }}});
     auto offsets_f = co_await ss::coroutine::as_future(
-      _client->list_offsets(model::schema_registry_internal_tp));
+      _client->list_offsets(std::move(req)));
     if (offsets_f.failed()) {
         rethrow_partition_error(offsets_f.get_exception());
     }

src/v/pandaproxy/schema_registry/kafka_client_transport.h

@@ -41,6 +41,7 @@ class kafka_client_transport final : public transport {
 
     ss::future<produce_result> produce(model::record_batch batch) override;
     ss::future<model::offset> get_high_watermark() override;
+    ss::future<model::offset> get_log_start() override;
     ss::future<> consume_range(
       model::offset start,
       model::offset end,
@@ -56,6 +57,7 @@ class kafka_client_transport final : public transport {
     bool has_ephemeral_credentials() const;
 
 private:
+    ss::future<model::offset> list_offset(model::timestamp timestamp);
     ss::future<> mitigate_error(std::exception_ptr eptr);
     ss::future<> inform(model::node_id);
     ss::future<> do_inform(model::node_id);

src/v/pandaproxy/schema_registry/rpc_transport.cc

@@ -47,6 +47,10 @@ throw_as_kafka_error(std::string_view context, cluster::errc ec) {
         throw kafka::exception(
           kafka::error_code::request_timed_out,
           fmt::format("{}: {}", context, ec));
+    case cluster::errc::offset_out_of_range:
+        throw kafka::exception(
+          kafka::error_code::offset_out_of_range,
+          fmt::format("{}: {}", context, ec));
     default:
         throw kafka::exception(
           kafka::error_code::unknown_server_error,
@@ -81,6 +85,19 @@ ss::future<model::offset> rpc_transport::get_high_watermark() {
     co_return kafka::offset_cast(result.value().high_watermark);
 }
 
+ss::future<model::offset> rpc_transport::get_log_start() {
+    auto result = co_await _client.get_single_partition_offsets(
+      model::schema_registry_internal_tp);
+    if (result.has_error()) {
+        throw_as_kafka_error(
+          "RPC get_partition_offsets failed", result.error());
+    }
+    // A broker that predates the start_offset field can't tell us where the
+    // log begins. Assume the whole log is available.
+    co_return kafka::offset_cast(
+      result.value().start_offset.value_or(kafka::offset{0}));
+}
+
 ss::future<> rpc_transport::consume_range(
   model::offset start,
   model::offset end,

src/v/pandaproxy/schema_registry/rpc_transport.h

@@ -33,6 +33,7 @@ class rpc_transport final : public transport {
 
     ss::future<produce_result> produce(model::record_batch batch) override;
     ss::future<model::offset> get_high_watermark() override;
+    ss::future<model::offset> get_log_start() override;
     ss::future<> consume_range(
       model::offset start,
       model::offset end,