kafka/server: remove foreign_ptr from response_ptr

The response object is always created and destroyed on the connection's
shard. The foreign_ptr wrapper was a historical artifact or oversight,
since in general a response may contain data from many shards so a
single-response level foreign pointer doesn't work. Today cross shard
deletion is handled by things _inside_ the response, e.g.,
by foreign_record_batch_readers nested somewhere inside the response.

The respond() path in request_context always serializes into fresh local
buffers, so the response iobuf never contains foreign fragments.

Replace foreign_ptr with a plain unique_ptr and add an oncore_auto
member to response to assert same-shard destruction in debug builds.

Author: travisdowns

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;
+    [[no_unique_address]] 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(