sr: Introduce transport interface

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/pandaproxy/schema_registry/BUILD

@@ -66,6 +66,19 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "transport",
+    hdrs = [
+        "transport.h",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/v/cluster",
+        "//src/v/model",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "core",
     srcs = [

src/v/pandaproxy/schema_registry/fwd.h

@@ -22,5 +22,6 @@ class seq_writer;
 class service;
 class sharded_store;
 class store;
+class transport;
 
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/transport.h

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2025 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include "cluster/errc.h"
+#include "cluster/types.h"
+#include "model/fundamental.h"
+#include "model/record.h"
+
+#include <seastar/core/coroutine.hh>
+#include <seastar/core/future.hh>
+#include <seastar/util/noncopyable_function.hh>
+
+#include <optional>
+
+namespace pandaproxy::schema_registry {
+
+/// Result of a produce operation — includes offset for collision detection.
+struct produce_result {
+    model::offset base_offset;
+};
+
+/// Abstract transport for schema registry's internal topic I/O.
+///
+/// Implementations exist for kafka::client (legacy) and kafka::data::rpc
+/// (RPC-based, no auth overhead).
+class transport {
+public:
+    transport() = default;
+    virtual ~transport() = default;
+    transport(const transport&) = delete;
+    transport& operator=(const transport&) = delete;
+    transport(transport&&) = delete;
+    transport& operator=(transport&&) = delete;
+
+    virtual ss::future<> stop() = 0;
+
+    /// Produce a batch to the _schemas topic. Returns the base_offset.
+    virtual ss::future<produce_result> produce(model::record_batch batch) = 0;
+
+    /// Get the high watermark (next offset) for the _schemas topic.
+    virtual ss::future<model::offset> get_high_watermark() = 0;
+
+    /// Consume batches from [start, end) on the _schemas topic.
+    /// Calls consumer(batch) for each batch. Handles pagination internally.
+    virtual ss::future<> consume_range(
+      model::offset start,
+      model::offset end,
+      ss::noncopyable_function<ss::future<>(model::record_batch)> consumer) = 0;
+
+    /// One-time startup configuration (credentials, ACLs).
+    /// Default: no-op (RPC transport needs no configuration).
+    virtual ss::future<> configure() { co_return; }
+
+    /// Create the internal schema registry topic.
+    virtual ss::future<cluster::errc> create_topic(
+      model::topic_namespace_view,
+      int32_t partition_count,
+      cluster::topic_properties,
+      std::optional<int16_t> replication_factor = std::nullopt) = 0;
+
+    /// Whether the transport uses ephemeral credentials.
+    virtual bool has_ephemeral_credentials() const = 0;
+};
+
+} // namespace pandaproxy::schema_registry