[v25.3.x] transform: multi-produce failure wedge fix

src/v/transform/BUILD

@@ -95,6 +95,7 @@ redpanda_cc_library(
         "//src/v/model:batch_compression",
         "//src/v/random:time_jitter",
         "//src/v/rpc",
+        "//src/v/ssx:abort_source",
         "//src/v/ssx:future_util",
         "//src/v/ssx:sformat",
         "//src/v/utils:backoff_policy",

src/v/transform/tests/test_fixture.cc

@@ -28,6 +28,11 @@ using namespace std::chrono_literals;
 namespace transform::testing {
 ss::future<> fake_sink::write(ss::chunked_fifo<model::record_batch> batches) {
     co_await _cork.wait();
+    if (_fail) {
+        throw std::runtime_error(
+          "failure to produce transform data: Current node is not a leader "
+          "for partition");
+    }
     for (auto& batch : batches) {
         for (auto& r : batch.copy_records()) {
             _records.push_back(std::move(r));
@@ -36,6 +41,16 @@ ss::future<> fake_sink::write(ss::chunked_fifo<model::record_batch> batches) {
     _cond_var.broadcast();
 }
 
+void fake_sink::fail_writes() {
+    _fail = true;
+    _cond_var.broadcast();
+}
+
+void fake_sink::resume_writes() {
+    _fail = false;
+    _cond_var.broadcast();
+}
+
 class read_timed_out : public ss::condition_variable_timed_out {
     const char* what() const noexcept override {
         return "waiting for read timed out";

src/v/transform/tests/test_fixture.h

@@ -108,10 +108,24 @@ class fake_sink : public sink {
      */
     void uncork();
 
+    /**
+     * Make every subsequent call to `write` throw, mimicking the real
+     * rpc_client_sink failing to produce (e.g. "Current node is not a leader
+     * for partition").
+     */
+    void fail_writes();
+
+    /**
+     * Stop failing writes, mimicking the transient produce failure (e.g. a
+     * leadership transfer) resolving so the sink can make progress again.
+     */
+    void resume_writes();
+
 private:
     ss::chunked_fifo<model::record> _records;
     ss::condition_variable _cond_var;
     ssx::semaphore _cork = {ssx::semaphore::max_counter(), "fake_sink"};
+    bool _fail = false;
 };
 
 class fake_offset_tracker : public offset_tracker {

src/v/transform/tests/transform_processor_test.cc

@@ -222,6 +222,13 @@ class ProcessorTestFixture : public ::testing::TestWithParam<fixture_param> {
 
     void cork_sink(model::output_topic_index idx) { _sinks[idx()]->cork(); }
     void uncork_sink(model::output_topic_index idx) { _sinks[idx()]->uncork(); }
+    void fail_sink(model::output_topic_index idx) {
+        _sinks[idx()]->fail_writes();
+    }
+    void recover_sink(model::output_topic_index idx) {
+        _sinks[idx()]->resume_writes();
+    }
+    bool processor_running() const { return _p->is_running(); }
 
     std::vector<model::output_topic_index> output_topics() const {
         std::vector<model::output_topic_index> indexes;
@@ -364,6 +371,95 @@ TEST_P(ProcessorTestFixture, LagOverflowBug) {
     EXPECT_EQ(lag(), 0);
 }
 
+// Regression test for the "stuck transform" bug. When one output's producer
+// fails to produce (e.g. a transient "not a leader for partition"), the failure
+// must be reported as state::errored so the manager restarts the processor,
+// regardless of how many outputs the transform has.
+//
+// The bug: with MULTIPLE outputs, `run_all_producers()` fanned out via
+// `ss::parallel_for_each()`, which captures the first exception but waits for
+// every loop to finish before resolving. Nothing aborted the shared abort
+// source, so the surviving producer loop(s) spin forever, pinning
+// parallel_for_each pending: state::errored never fired, the processor stayed
+// "running", and it made no progress until an external stop (== an `rpk
+// transform pause`/`resume`). Single-output transforms were unaffected because
+// parallel_for_each over one element resolves exceptionally right away.
+//
+// The fix drives the producer loops off a composite abort source, so the first
+// failing producer unwinds its siblings (without touching the processor's own
+// abort source) and propagates exactly one error.
+TEST_P(ProcessorTestFixture, ProduceFailureIsReportedForAnyOutputCount) {
+    set_tee_output();
+
+    // Healthy baseline so the pipeline is flowing and committed.
+    auto baseline = make_records(1);
+    push_batch(baseline);
+    for (auto o : output_topics()) {
+        EXPECT_THAT(read_records(o, 1), SameRecords(baseline));
+    }
+    ASSERT_TRUE(wait_for_all_committed());
+
+    // Fail only output 0; any other outputs stay healthy.
+    fail_sink(model::output_topic_index(0));
+    push_batch(make_records(1));
+    tests::drain_task_queue().get();
+
+    // The producers all unwind instead of wedging. The processor still reports
+    // running (its abort source is untouched); like a single-output failure, it
+    // relies on the manager observing the error and restarting it.
+    EXPECT_TRUE(processor_running());
+    // Output 0's producer died, so nothing is ever written there.
+    EXPECT_TRUE(sink_empty(model::output_topic_index(0)));
+    // The failure is reported exactly once, no matter the output count: the
+    // consumer/transform loops keep running on the untouched abort source, so
+    // the manager restarts the processor without spurious duplicate errors.
+    EXPECT_EQ(error_count(), 1u)
+      << "produce failure should be reported as errored exactly once";
+}
+
+// A produce failure must leave the processor cleanly restartable: the manager's
+// recovery path stops and then restarts the same processor instance, so stop()
+// has to tear down (and the wedge fix must not have aborted the processor's own
+// abort source, which would short-circuit stop() and leave the engine started
+// for the restart to double-start). Exercise that full cycle and assert the
+// restart is clean and the processor resumes producing.
+TEST_P(ProcessorTestFixture, RecoversFromProduceFailureViaRestart) {
+    set_tee_output();
+
+    // Healthy baseline so the pipeline is flowing and committed.
+    auto baseline = make_records(1);
+    push_batch(baseline);
+    for (auto o : output_topics()) {
+        EXPECT_THAT(read_records(o, 1), SameRecords(baseline));
+    }
+    ASSERT_TRUE(wait_for_all_committed());
+
+    // A producer fails and the error is reported, exactly what the manager
+    // observes before it restarts.
+    fail_sink(model::output_topic_index(0));
+    push_batch(make_records(1));
+    tests::drain_task_queue().get();
+    ASSERT_EQ(error_count(), 1u);
+
+    // Recovery: the transient failure clears and the manager stops and restarts
+    // the same processor instance.
+    recover_sink(model::output_topic_index(0));
+    restart();
+    tests::drain_task_queue().get();
+
+    // A clean restart produces no further errors.
+    ASSERT_EQ(error_count(), 1u)
+      << "restart after a produce failure should not report a new error";
+    EXPECT_TRUE(processor_running());
+
+    // The restarted processor resumes producing to every output.
+    auto resumed = make_records(1);
+    push_batch(resumed);
+    for (auto o : output_topics()) {
+        EXPECT_FALSE(read_records(o, 1).empty());
+    }
+}
+
 INSTANTIATE_TEST_SUITE_P(
   GenericProcessorTest,
   ProcessorTestFixture,

src/v/transform/transform_processor.cc

@@ -18,6 +18,7 @@
 #include "model/timestamp.h"
 #include "model/transform.h"
 #include "random/simple_time_jitter.h"
+#include "ssx/abort_source.h"
 #include "ssx/future-util.h"
 #include "wasm/engine.h"
 
@@ -343,19 +344,50 @@ ss::future<> processor::run_transform_loop() {
 ss::future<> processor::run_all_producers(
   absl::flat_hash_map<model::output_topic_index, kafka::offset>
     latest_committed) {
-    return ss::parallel_for_each(
-      _outputs, [this, committed = std::move(latest_committed)](auto& entry) {
+    // parallel_for_each captures the first exception but waits for every loop
+    // to finish before resolving, and a producer loop only exits when its abort
+    // source fires. Drive the loops off a composite source that aborts when
+    // either the processor stops (_as) or a producer fails (local_producer_as,
+    // our "a producer failed" signal), so the first failure unwinds its
+    // siblings promptly instead of leaving them spinning forever.
+    ss::abort_source local_producer_as;
+    ssx::composite_abort_source producers_as(_as, local_producer_as);
+
+    std::exception_ptr failure;
+    co_await ss::parallel_for_each(
+      _outputs,
+      [this,
+       &failure,
+       &local_producer_as,
+       &producers_as,
+       committed = std::move(latest_committed)](auto& entry) {
           output& out = entry.second;
           return run_producer_loop(
-            out.index, &out.queue, out.sink.get(), committed.at(out.index));
+                   out.index,
+                   &out.queue,
+                   out.sink.get(),
+                   committed.at(out.index),
+                   producers_as.as())
+            .handle_exception([&failure,
+                               &local_producer_as](std::exception_ptr ep) {
+                if (!local_producer_as.abort_requested()) {
+                    failure = std::move(ep);
+                    local_producer_as.request_abort_ex(
+                      std::make_exception_ptr(processor_shutdown_exception()));
+                }
+            });
       });
+    if (failure) {
+        std::rethrow_exception(failure);
+    }
 }
 
 ss::future<> processor::run_producer_loop(
   model::output_topic_index index,
   transfer_queue<transformed_output>* queue,
   sink* sink,
-  kafka::offset last_committed) {
+  kafka::offset last_committed,
+  ss::abort_source& as) {
     vlog(
       _logger.debug,
       "starting producer {} - last committed: {}",
@@ -365,8 +397,8 @@ ss::future<> processor::run_producer_loop(
     // to suppress records until we've reached the previous offset we've
     // committed.
     bool suppress = true;
-    while (!_as.abort_requested()) {
-        auto popped = co_await queue->pop_all(&_as);
+    while (!as.abort_requested()) {
+        auto popped = co_await queue->pop_all(&as);
         if (popped.empty()) {
             continue;
         }

src/v/transform/transform_processor.h

@@ -114,7 +114,8 @@ class processor {
       model::output_topic_index,
       transfer_queue<transformed_output>*,
       sink*,
-      kafka::offset);
+      kafka::offset,
+      ss::abort_source&);
     ss::future<> poll_sleep();
     ss::future<absl::flat_hash_map<model::output_topic_index, kafka::offset>>
     load_latest_committed();