schema_registry: add schema_registry_replay_on_startup

Recovery of the _schemas store is lazy: nothing loads the topic until the
first request (or an internal schema::registry access) reaches
ensure_started(). On a large registry the first request then blocks for the
whole replay, and there is no way to hydrate the store ahead of traffic.

Add a cluster config (default off) that, when set, proactively drives the
replay on the reader shard at service start-up, fire-and-forget under the
gate so it does not block broker start-up. It reuses the single-flight
ensure_started() path, so a request that races the eager trigger still
results in exactly one replay.

Covered by ducktape tests for both settings: with the config set the store
hydrates on restart with no client request, and with it unset recovery stays
lazy until the first request.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: sjust-redpanda

src/v/config/configuration.cc

@@ -3974,6 +3974,15 @@ configuration::configuration()
        .visibility = visibility::user,
        .aliases = {"schema_registry_normalize_on_startup"}},
       false)
+  , schema_registry_replay_on_startup(
+      *this,
+      "schema_registry_replay_on_startup",
+      "Replay the internal `_schemas` topic into the Schema Registry store at "
+      "broker start-up instead of lazily on the first request. Makes recovery "
+      "time predictable and keeps the first request from blocking behind a "
+      "full replay.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::user},
+      false)
   , schema_registry_avro_use_named_references(
       *this, "schema_registry_avro_use_named_references")
   , schema_registry_enable_qualified_subjects(

src/v/config/configuration.h

@@ -683,6 +683,7 @@ struct configuration final : public config_store {
 
     enterprise<property<bool>> schema_registry_enable_authorization;
     property<bool> schema_registry_always_normalize;
+    property<bool> schema_registry_replay_on_startup;
     deprecated_property schema_registry_avro_use_named_references;
     property<bool> schema_registry_enable_qualified_subjects;
     bounded_property<size_t> schema_registry_sync_memory_bytes;

src/v/pandaproxy/schema_registry/BUILD

@@ -477,6 +477,7 @@ redpanda_cc_library(
         "//src/v/security:request_auth",
         "//src/v/security/audit",
         "//src/v/security/audit:types",
+        "//src/v/ssx:future_util",
         "//src/v/ssx:semaphore",
         "//src/v/ssx:sformat",
         "//src/v/strings:string_switch",

src/v/pandaproxy/schema_registry/service.cc

@@ -36,6 +36,7 @@
 #include "security/audit/audit_log_manager.h"
 #include "security/authorizer.h"
 #include "security/request_auth.h"
+#include "ssx/future-util.h"
 #include "ssx/semaphore.h"
 #include "utils/tristate.h"
 
@@ -824,10 +825,30 @@ service::service(
 ss::future<> service::start() {
     static std::vector<model::broker_endpoint> not_advertised{};
     _server.routes(get_schema_registry_routes(_gate, _ensure_started));
-    co_return co_await _server.start(
+    co_await _server.start(
       _config.schema_registry_api(),
       _config.schema_registry_api_tls(),
       not_advertised);
+
+    if (
+      ss::this_shard_id() == seq_writer::reader_shard
+      && config::shard_local_cfg().schema_registry_replay_on_startup()) {
+        // Proactively hydrate the store rather than wait for the first request.
+        // Fire-and-forget under the gate: a large topic can take a while to
+        // replay and must not block broker start-up. This goes through the
+        // single-flight ensure_started(), so a request that races it still
+        // triggers exactly one replay.
+        ssx::spawn_with_gate(_gate, [this] {
+            return ensure_started().handle_exception(
+              [](const std::exception_ptr& e) {
+                  vlog(
+                    srlog.warn,
+                    "eager _schemas replay failed at start-up; will retry on "
+                    "the first request: {}",
+                    e);
+              });
+        });
+    }
 }
 
 ss::future<> service::stop() {

tests/rptest/tests/schema_registry_test.py

@@ -12134,3 +12134,108 @@ def hammer():
         )
         assert r.status_code == requests.codes.ok, r.text
         assert len(r.json()) >= self.N_SUBJECTS
+
+    @cluster(num_nodes=3, log_allow_list=_SR_STARTUP_RECOVERY_LOG_ALLOW_LIST)
+    def test_replay_on_startup_without_request(self):
+        """
+        With schema_registry_replay_on_startup enabled, the store must hydrate
+        proactively at startup without any client request driving it.
+
+        admin restart_service re-runs service::start() (via api::restart ->
+        start -> service::start), which is where the eager trigger lives, and
+        does not restart the broker process, so the log persists and we can
+        watch for the replay without touching the SR API.
+        """
+        node = self.redpanda.nodes[0]
+        host = node.account.hostname
+
+        # Populate _schemas so a restart has a topic to replay.
+        self._register_schemas(host)
+        wait_until(
+            lambda: self._all_subjects_served(host),
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="subjects not visible after registration",
+        )
+
+        self.redpanda.set_cluster_config({"schema_registry_replay_on_startup": True})
+
+        init_before = self.redpanda.count_log_node(node, self.INIT_MARKER)
+
+        admin = Admin(self.redpanda)
+        result = admin.restart_service(rp_service="schema-registry", node=node)
+        assert result.status_code == requests.codes.ok, (
+            f"restart_service failed: {result.status_code} {result.text}"
+        )
+
+        # Deliberately issue NO SR request. The eager startup trigger must drive
+        # the replay on its own. count_log_node greps the broker log over ssh,
+        # so polling here does not hit the SR API and cannot trigger the lazy
+        # path.
+        wait_until(
+            lambda: self.redpanda.count_log_node(node, self.INIT_MARKER) > init_before,
+            timeout_sec=60,
+            backoff_sec=1,
+            err_msg="eager startup replay did not run without any request",
+        )
+
+        replays = self.redpanda.count_log_node(node, self.INIT_MARKER) - init_before
+        assert replays == 1, (
+            f"expected exactly one eager replay, saw {replays} "
+            f"('{self.INIT_MARKER}' delta)."
+        )
+
+        # The store hydrated without a request; a request now succeeds
+        # immediately rather than blocking on a cold replay.
+        assert self._all_subjects_served(host)
+
+    @cluster(num_nodes=3, log_allow_list=_SR_STARTUP_RECOVERY_LOG_ALLOW_LIST)
+    def test_no_replay_on_startup_by_default(self):
+        """
+        With schema_registry_replay_on_startup unset (the default), recovery
+        stays lazy: a restart does not replay _schemas until a request (or an
+        internal access) arrives. The complement of
+        test_replay_on_startup_without_request, and a guard that the config
+        actually gates the eager path.
+        """
+        node = self.redpanda.nodes[0]
+        host = node.account.hostname
+
+        self._register_schemas(host)
+        wait_until(
+            lambda: self._all_subjects_served(host),
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="subjects not visible after registration",
+        )
+
+        # Deliberately leave schema_registry_replay_on_startup at its default
+        # (off). do_start() logs INIT_MARKER once per run, so a flat count after
+        # a restart with no request means recovery did not run.
+        init_before = self.redpanda.count_log_node(node, self.INIT_MARKER)
+
+        admin = Admin(self.redpanda)
+        result = admin.restart_service(rp_service="schema-registry", node=node)
+        assert result.status_code == requests.codes.ok, (
+            f"restart_service failed: {result.status_code} {result.text}"
+        )
+
+        # No SR request is issued. Give recovery ample time to (not) happen;
+        # a real replay of this topic completes in well under a second.
+        time.sleep(15)
+        assert self.redpanda.count_log_node(node, self.INIT_MARKER) == init_before, (
+            "recovery ran at startup without the config set and without a "
+            "request; it should be lazy by default"
+        )
+
+        # The first request now lazily triggers exactly one replay and serves.
+        wait_until(
+            lambda: self._all_subjects_served(host),
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="lazy recovery did not serve after the first request",
+        )
+        replays = self.redpanda.count_log_node(node, self.INIT_MARKER) - init_before
+        assert replays == 1, (
+            f"expected exactly one lazy replay after the first request, saw {replays}."
+        )