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

@@ -145,32 +145,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",
@@ -183,7 +170,6 @@ redpanda_cc_library(
         "@boost//:algorithm",
         "@boost//:graph",
         "@boost//:math",
-        "@boost//:multi_index",
         "@boost//:outcome",
         "@boost//:range",
         "@fmt",
@@ -201,15 +187,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",
     ],
 )
@@ -248,6 +245,7 @@ redpanda_cc_library(
     ],
     implementation_deps = [
         ":core",
+        ":kafka_client_transport",
         "//src/v/base",
         "//src/v/bytes:iobuf_parser",
         "//src/v/cluster",
@@ -266,9 +264,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",
@@ -305,6 +301,7 @@ redpanda_cc_library(
         ":config",
         ":rjson",
         ":subject_name_strategy",
+        ":transport",
         ":types",
         "//src/v/cluster:client_quota_backend",
         "//src/v/cluster:cluster_link_table",
@@ -316,7 +313,6 @@ redpanda_cc_library(
         "//src/v/cluster:plugin_table",
         "//src/v/cluster:rpc_utils",
         "//src/v/config:startup_config",
-        "//src/v/kafka/client",
     ],
 )
 
@@ -345,6 +341,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,38 +84,36 @@ 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] { return std::ref(_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] { return std::ref(_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));
 
+    co_await _transport.invoke_on_all(&kafka_client_transport::configure);
     co_await _service.invoke_on_all(&service::start);
 
     if (ss::this_shard_id() == 0) {
@@ -143,10 +142,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) {
@@ -171,7 +170,7 @@ const kafka::client::configuration& api::get_client_config() const {
 }
 
 bool api::has_ephemeral_credentials() const {
-    return _service.local().has_ephemeral_credentials();
+    return _transport.local().has_ephemeral_credentials();
 }
 
 ss::future<> api::contribute_metrics(

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,10 @@ 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);
+                  return seq._transport->consume_range(
+                    seq._loaded_offset + model::offset{1},
+                    offset + model::offset{1},
+                    consume_to_store{seq._store, seq});
               } else {
                   vlog(srlog.trace, "wait_for clean (offset  {})", offset);
                   return ss::make_ready_future<>();
@@ -174,29 +160,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,11 +48,11 @@ 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)) {}
@@ -92,7 +92,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;