sr: Swap in transport for kafka::client in service & seq_writer

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

Author: oleiman

src/v/pandaproxy/schema_registry/BUILD

@@ -143,32 +143,19 @@ redpanda_cc_library(
         "util.h",
     ],
     implementation_deps = [
-        "//src/v/bytes:iobuf_parser",
         "//src/v/bytes:streambuf",
         "//src/v/cluster",
         "//src/v/container:chunked_hash_map",
-        "//src/v/container:chunked_vector",
         "//src/v/container:json",
         "//src/v/hashing:jump_consistent",
         "//src/v/hashing:xx",
-        "//src/v/json",
         "//src/v/kafka/protocol",
-        "//src/v/metrics",
-        "//src/v/model",
-        "//src/v/model:batch_compression",
         "//src/v/pandaproxy:core",
-        "//src/v/pandaproxy:logger",
-        "//src/v/random:time_jitter",
         "//src/v/schema/protobuf:confluent_type_cc_proto",
         "//src/v/schema/protobuf:google_type_cc_proto",
         "//src/v/ssx:future_util",
-        "//src/v/ssx:semaphore",
-        "//src/v/ssx:sformat",
-        "//src/v/storage:record_batch_builder",
-        "//src/v/strings:string_switch",
         "//src/v/utils:base64",
         "//src/v/utils:named_type",
-        "//src/v/utils:retry",
         "//src/v/utils:to_string",
         "@abseil-cpp//absl/algorithm:container",
         "@abseil-cpp//absl/container:btree",
@@ -181,7 +168,6 @@ redpanda_cc_library(
         "@boost//:algorithm",
         "@boost//:graph",
         "@boost//:math",
-        "@boost//:multi_index",
         "@boost//:outcome",
         "@boost//:range",
         "@fmt",
@@ -199,15 +185,26 @@ redpanda_cc_library(
         ":config",
         ":rjson",
         ":subject_name_strategy",
+        ":transport",
         ":types",
         "//src/v/base",
+        "//src/v/bytes:iobuf_parser",
         "//src/v/config",
         "//src/v/config:startup_config",
-        "//src/v/kafka/client",
-        "//src/v/kafka/client:config_utils",
-        "//src/v/kafka/client:configuration",
-        "//src/v/kafka/client:exceptions",
+        "//src/v/container:chunked_vector",
+        "//src/v/json",
+        "//src/v/metrics",
+        "//src/v/model",
+        "//src/v/model:batch_compression",
         "//src/v/pandaproxy:json",
+        "//src/v/pandaproxy:logger",
+        "//src/v/random:time_jitter",
+        "//src/v/ssx:semaphore",
+        "//src/v/ssx:sformat",
+        "//src/v/storage:record_batch_builder",
+        "//src/v/strings:string_switch",
+        "//src/v/utils:retry",
+        "@boost//:multi_index",
         "@seastar",
     ],
 )
@@ -246,6 +243,7 @@ redpanda_cc_library(
     ],
     implementation_deps = [
         ":core",
+        ":kafka_client_transport",
         "//src/v/base",
         "//src/v/bytes:iobuf_parser",
         "//src/v/cluster",
@@ -264,9 +262,7 @@ redpanda_cc_library(
         "//src/v/container:chunked_vector",
         "//src/v/container:json",
         "//src/v/json",
-        "//src/v/kafka/client:config_utils",
         "//src/v/kafka/client:configuration",
-        "//src/v/kafka/client:exceptions",
         "//src/v/kafka/data/rpc",
         "//src/v/kafka/protocol",
         "//src/v/kafka/protocol:create_topics",
@@ -303,6 +299,7 @@ redpanda_cc_library(
         ":config",
         ":rjson",
         ":subject_name_strategy",
+        ":transport",
         ":types",
         "//src/v/cluster:client_quota_backend",
         "//src/v/cluster:cluster_link_table",
@@ -314,7 +311,6 @@ redpanda_cc_library(
         "//src/v/cluster:plugin_table",
         "//src/v/cluster:rpc_utils",
         "//src/v/config:startup_config",
-        "//src/v/kafka/client",
     ],
 )
 
@@ -343,6 +339,7 @@ redpanda_cc_library(
     visibility = ["//visibility:public"],
     deps = [
         ":server",
+        ":transport",
         "//src/v/config:startup_config",
         "//src/v/pandaproxy:core",
     ],

src/v/pandaproxy/schema_registry/api.cc

@@ -16,6 +16,7 @@
 #include "kafka/data/rpc/deps.h"
 #include "pandaproxy/logger.h"
 #include "pandaproxy/schema_registry/configuration.h"
+#include "pandaproxy/schema_registry/kafka_client_transport.h"
 #include "pandaproxy/schema_registry/schema_id_cache.h"
 #include "pandaproxy/schema_registry/service.h"
 #include "pandaproxy/schema_registry/sharded_store.h"
@@ -83,35 +84,34 @@ ss::future<> api::start() {
         return config::shard_local_cfg()
           .kafka_schema_id_validation_cache_capacity.bind();
     }));
-    co_await _client.start(
-      config::to_yaml(_client_cfg, config::redact_secrets::no),
-      [this](std::exception_ptr ex) {
-          return _service.local().mitigate_error(ex);
-      });
+    co_await _transport.start(
+      std::ref(_client_cfg),
+      std::ref(*_controller),
+      ss::sharded_parameter([this]() {
+          return kafka::data::rpc::topic_creator::make_default(
+            _controller.get());
+      }));
     co_await _sequencer.start(
       _node_id,
       _sg,
-      std::ref(_client),
+      ss::sharded_parameter(
+        [this]() -> transport* { return &_transport.local(); }),
       std::ref(*_store),
       ss::sharded_parameter([this] {
           return std::make_unique<sequence_state_checker_impl>(_controller);
       }));
     co_await _service.start(
       config::to_yaml(_cfg, config::redact_secrets::no),
-      config::to_yaml(_client_cfg, config::redact_secrets::no),
       _sg,
       _max_memory,
-      std::ref(_client),
+      ss::sharded_parameter(
+        [this]() -> transport* { return &_transport.local(); }),
       std::ref(*_store),
       std::ref(_sequencer),
       ss::sharded_parameter([this]() {
           return kafka::data::rpc::topic_metadata_cache::make_default(
             _metadata_cache);
       }),
-      ss::sharded_parameter([this]() {
-          return kafka::data::rpc::topic_creator::make_default(
-            _controller.get());
-      }),
       std::ref(_controller),
       std::ref(_audit_mgr));
 
@@ -143,10 +143,10 @@ ss::future<> api::stop() {
         // Reset gate to support api restart
         _metrics_gate = ss::gate{};
     }
-    co_await _client.invoke_on_all(&kafka::client::client::stop);
+    co_await _transport.invoke_on_all(&kafka_client_transport::stop);
     co_await _service.stop();
     co_await _sequencer.stop();
-    co_await _client.stop();
+    co_await _transport.stop();
     co_await _schema_id_cache.stop();
     co_await _schema_id_validation_probe.stop();
     if (_store) {

src/v/pandaproxy/schema_registry/api.h

@@ -13,7 +13,7 @@
 
 #include "base/seastarx.h"
 #include "cluster/metrics_reporter.h"
-#include "kafka/client/fwd.h"
+#include "kafka/client/configuration.h"
 #include "model/metadata.h"
 #include "pandaproxy/schema_registry/fwd.h"
 #include "security/fwd.h"
@@ -73,7 +73,7 @@ class api {
     ss::sharded<cluster::metadata_cache>* _metadata_cache;
     std::unique_ptr<cluster::controller>& _controller;
 
-    ss::sharded<kafka::client::client> _client;
+    ss::sharded<kafka_client_transport> _transport;
     std::unique_ptr<pandaproxy::schema_registry::sharded_store> _store;
     ss::sharded<schema_id_validation_probe> _schema_id_validation_probe;
     ss::sharded<schema_id_cache> _schema_id_cache;

src/v/pandaproxy/schema_registry/fwd.h

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

src/v/pandaproxy/schema_registry/seq_writer.cc

@@ -11,14 +11,14 @@
 
 #include "base/vassert.h"
 #include "base/vlog.h"
-#include "kafka/client/client_fetch_batch_reader.h"
 #include "model/namespace.h"
 #include "pandaproxy/logger.h"
 #include "pandaproxy/schema_registry/error.h"
 #include "pandaproxy/schema_registry/errors.h"
 #include "pandaproxy/schema_registry/exceptions.h"
 #include "pandaproxy/schema_registry/sharded_store.h"
 #include "pandaproxy/schema_registry/storage.h"
+#include "pandaproxy/schema_registry/transport.h"
 #include "pandaproxy/schema_registry/types.h"
 #include "storage/record_batch_builder.h"
 
@@ -106,17 +106,7 @@ struct batch_builder : public storage::record_batch_builder {
 /// a REST API endpoint that requires global knowledge of latest
 /// data (i.e. any listings)
 ss::future<> seq_writer::read_sync() {
-    auto offsets = co_await _client.local().list_offsets(
-      model::schema_registry_internal_tp);
-    if (
-      offsets.data.topics.size() != 1
-      || offsets.data.topics[0].partitions.size() != 1) {
-        throw kafka::exception(
-          kafka::error_code::unknown_server_error,
-          "Malformed ListOffsets Kafka response for internal topic");
-    }
-
-    auto max_offset = offsets.data.topics[0].partitions[0].offset;
+    auto max_offset = co_await _transport->get_high_watermark();
     co_await wait_for(max_offset - model::offset{1});
     co_await _store.process_marked_schemas();
 }
@@ -145,14 +135,12 @@ ss::future<> seq_writer::wait_for(model::offset offset) {
                     "wait_for dirty!  Reading {}..{}",
                     seq._loaded_offset,
                     offset);
-
-                  return kafka::client::make_client_fetch_batch_reader(
-                           seq._client.local(),
-                           model::schema_registry_internal_tp,
-                           seq._loaded_offset + model::offset{1},
-                           offset + model::offset{1})
-                    .consume(
-                      consume_to_store{seq._store, seq}, model::no_timeout);
+                  consume_to_store c{seq._store, seq};
+                  return seq._transport->consume_range(
+                    seq._loaded_offset + model::offset{1},
+                    offset + model::offset{1},
+                    [c = std::move(c)](this auto, model::record_batch batch)
+                      -> ss::future<> { co_await c(std::move(batch)); });
               } else {
                   vlog(srlog.trace, "wait_for clean (offset  {})", offset);
                   return ss::make_ready_future<>();
@@ -174,29 +162,25 @@ ss::future<bool> seq_writer::produce_and_apply(
       write_at.value_or(batch.base_offset()) == batch.base_offset(),
       "Set the base_offset to the expected write_at");
 
-    kafka::partition_produce_response res
-      = co_await _client.local().produce_record_batch(
-        model::schema_registry_internal_tp, batch.copy());
-
-    if (res.error_code != kafka::error_code::none) {
-        throw kafka::exception(res.error_code, res.error_message.value_or(""));
-    }
+    auto result = co_await _transport->produce(batch.copy());
 
-    auto success = write_at.value_or(res.base_offset) == res.base_offset;
+    auto success = write_at.value_or(result.base_offset) == result.base_offset;
     if (success) {
         vlog(
-          srlog.debug, "seq_writer: Successful write at {}", res.base_offset);
+          srlog.debug,
+          "seq_writer: Successful write at {}",
+          result.base_offset);
         co_await consume_to_store(_store, *this)(std::move(batch));
         co_await _store.process_marked_schemas();
     } else {
         vlog(
           srlog.debug,
           "seq_writer: Failed write at {} (wrote at {})",
-          write_at,
-          res.base_offset);
+          write_at.value_or(model::offset{-1}),
+          result.base_offset);
     }
     co_return success;
-};
+}
 
 ss::future<> seq_writer::advance_offset(model::offset offset) {
     auto remote = [offset](seq_writer& s) { s.advance_offset_inner(offset); };

src/v/pandaproxy/schema_registry/seq_writer.h

@@ -10,12 +10,12 @@
 #pragma once
 
 #include "base/outcome.h"
-#include "kafka/client/client.h"
 #include "pandaproxy/logger.h"
 #include "pandaproxy/schema_registry/error.h"
 #include "pandaproxy/schema_registry/errors.h"
 #include "pandaproxy/schema_registry/exceptions.h"
 #include "pandaproxy/schema_registry/sharded_store.h"
+#include "pandaproxy/schema_registry/transport.h"
 #include "pandaproxy/schema_registry/types.h"
 #include "random/simple_time_jitter.h"
 #include "ssx/semaphore.h"
@@ -48,14 +48,16 @@ class seq_writer final : public ss::peering_sharded_service<seq_writer> {
     seq_writer(
       model::node_id node_id,
       ss::smp_service_group smp_group,
-      ss::sharded<kafka::client::client>& client,
+      transport* transport,
       sharded_store& store,
       std::unique_ptr<sequence_state_checker> state_checker)
       : _smp_opts(ss::smp_submit_to_options{smp_group})
-      , _client(client)
+      , _transport(transport)
       , _store(store)
       , _node_id(node_id)
-      , _state_checker(std::move(state_checker)) {}
+      , _state_checker(std::move(state_checker)) {
+        vassert(_transport != nullptr, "Transport unexpectedly null");
+    }
 
     ss::future<> read_sync();
 
@@ -92,7 +94,7 @@ class seq_writer final : public ss::peering_sharded_service<seq_writer> {
 private:
     ss::smp_submit_to_options _smp_opts;
 
-    ss::sharded<kafka::client::client>& _client;
+    transport* _transport;
     sharded_store& _store;
 
     model::node_id _node_id;