dl/translation/writer: propagate an abort source to create writer

Author: bharathv

src/v/datalake/data_writer_interface.h

@@ -174,7 +174,8 @@ class parquet_file_writer_factory {
 
     virtual ss::future<
       result<std::unique_ptr<parquet_file_writer>, writer_error>>
-    create_writer(const iceberg::struct_type& /* schema */) = 0;
+    create_writer(const iceberg::struct_type& /* schema */, ss::abort_source&)
+      = 0;
 };
 
 } // namespace datalake

src/v/datalake/local_parquet_file_writer.cc

@@ -175,7 +175,7 @@ local_parquet_file_writer_factory::local_parquet_file_writer_factory(
 
 ss::future<result<std::unique_ptr<parquet_file_writer>, writer_error>>
 local_parquet_file_writer_factory::create_writer(
-  const iceberg::struct_type& schema) {
+  const iceberg::struct_type& schema, ss::abort_source&) {
     auto writer = std::make_unique<local_parquet_file_writer>(
       create_filename(), _writer_factory, *_mem_tracker);
 

src/v/datalake/local_parquet_file_writer.h

@@ -68,7 +68,7 @@ class local_parquet_file_writer_factory : public parquet_file_writer_factory {
       std::unique_ptr<writer_mem_tracker>);
 
     ss::future<result<std::unique_ptr<parquet_file_writer>, writer_error>>
-    create_writer(const iceberg::struct_type& schema) final;
+    create_writer(const iceberg::struct_type& schema, ss::abort_source&) final;
 
 private:
     local_path create_filename() const;

src/v/datalake/partitioning_writer.cc

@@ -57,7 +57,7 @@ ss::future<writer_error> partitioning_writer::add_data(
     }
     auto writer_iter = writers_.find(pk);
     if (writer_iter == writers_.end()) {
-        auto writer_res = co_await writer_factory_.create_writer(type_);
+        auto writer_res = co_await writer_factory_.create_writer(type_, as);
         if (writer_res.has_error()) {
             vlog(
               datalake_log.error,

src/v/datalake/tests/test_data_writer.h

@@ -73,7 +73,8 @@ class test_data_writer_factory : public parquet_file_writer_factory {
       : _return_error{return_error} {}
 
     ss::future<result<std::unique_ptr<parquet_file_writer>, writer_error>>
-    create_writer(const iceberg::struct_type& schema) override {
+    create_writer(
+      const iceberg::struct_type& schema, ss::abort_source&) override {
         co_return std::make_unique<test_data_writer>(
           std::move(schema), _return_error);
     }
@@ -122,7 +123,8 @@ class test_serde_parquet_data_writer : public parquet_file_writer {
 class test_serde_parquet_writer_factory : public parquet_file_writer_factory {
 public:
     ss::future<result<std::unique_ptr<parquet_file_writer>, writer_error>>
-    create_writer(const iceberg::struct_type& schema) override {
+    create_writer(
+      const iceberg::struct_type& schema, ss::abort_source&) override {
         auto ostream_writer = co_await _serde_parquet_factory.create_writer(
           schema, utils::make_null_output_stream(), _mem_tracker);