kafka/server: remove foreign_ptr from response_ptr

src/v/base/oncore.h

@@ -48,3 +48,11 @@ class oncore final {
     do {                                                                       \
         expression_in_debug_mode((member).assert_shard_source_location());     \
     } while (0)
+
+/// An oncore that debug-asserts same-shard in its destructor.
+/// Embed as a member to catch cross-shard destruction in debug builds.
+// NOLINTNEXTLINE(cppcoreguidelines-special-member-functions,hicpp-special-member-functions)
+struct oncore_auto final {
+    ~oncore_auto() noexcept { oncore_debug_verify(_oncore); }
+    oncore _oncore;
+};

src/v/kafka/server/connection_context.h

@@ -19,6 +19,7 @@
 #include "kafka/server/handlers/details/security.h"
 #include "kafka/server/handlers/handler_probe.h"
 #include "kafka/server/logger.h"
+#include "kafka/server/response.h"
 #include "net/connection.h"
 #include "net/server_probe.h"
 #include "proto/redpanda/core/admin/v2/kafka_connections.proto.h"
@@ -49,8 +50,6 @@
 
 namespace kafka {
 
-using response_ptr = ss::foreign_ptr<std::unique_ptr<response>>;
-
 using closed_connections_t
   = std::deque<ss::lw_shared_ptr<const proto::admin::kafka_connection>>;
 

src/v/kafka/server/protocol_utils.cc

@@ -155,7 +155,8 @@ ss::scattered_message<char> response_as_scattered(response_ptr response) {
           msg.append_static(src, sz);
           return ss::stop_iteration::no;
       });
-    // MUST be the foreign ptr not the iobuf
+    // The response must outlive the scattered message since the message
+    // references the iobuf fragments directly via append_static.
     msg.on_delete([response = std::move(response)] {});
     return msg;
 }

src/v/kafka/server/response.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "base/oncore.h"
 #include "base/seastarx.h"
 #include "bytes/iobuf.h"
 #include "kafka/protocol/types.h"
@@ -63,9 +64,10 @@ class response {
     std::optional<tagged_fields> _tags;
     iobuf _buf;
     protocol::encoder _writer;
+    oncore_auto _oncore;
 };
 
-using response_ptr = ss::foreign_ptr<std::unique_ptr<response>>;
+using response_ptr = std::unique_ptr<response>;
 
 struct process_result_stages {
     process_result_stages(