Merge pull request #25946 from vbotbuildovich/backport-pr-25925-v25.1.x-321

[v25.1.x] [CORE-9906] datalake: fix translation double-stop

Author: piyushredpanda

src/v/datalake/translation/scheduling.h

@@ -231,7 +231,7 @@ class translator {
      * matter what stat it is in (e.g. running, waiting, idle). The translator
      * is free to clear the request after taking action.
      */
-    virtual void set_finish_translation() {}
+    virtual void set_finish_translation() = 0;
 
     /**
      * Return true if the translator is still in the progress of satisfying the

src/v/datalake/translation/scheduling_policies.cc

@@ -384,19 +384,18 @@ ss::future<> fair_scheduling_policy::on_resource_exhaustion(
                  > b.status().memory_bytes_reserved;
       });
 
-    auto num_running = executor.running.size();
     // pick the earliest scheduled translator and force a flush.
+    auto& executable = *executor.running.begin();
     vlog(
       datalake_log.debug,
       "[{}] stopping translator due to memory exhaustion",
-      *executor.running.begin());
-    executor.stop_translation(
-      *executor.running.begin(), translator::stop_reason::oom);
-
-    while (mem_tracker.memory_exhausted() && !executor.as.abort_requested()
-           && executor.running.size() == num_running) {
-        co_await ss::sleep_abortable(polling_interval, executor.as);
-    }
+      executable);
+    co_await finish_translator(
+      executor,
+      finish_choice_info(
+        executable.translator_ptr()->id(),
+        finish_choice_info::status::running,
+        translator::stop_reason::oom));
 }
 
 enum fair_scheduling_policy::finish_choice_info::status

src/v/datalake/translation/tests/scheduler_fixture.cc

@@ -15,6 +15,8 @@
 #include "ssx/future-util.h"
 #include "test_utils/randoms.h"
 
+#include <seastar/util/defer.hh>
+
 using namespace std::chrono_literals;
 
 namespace datalake::translation::scheduling {
@@ -163,6 +165,8 @@ ss::future<> mock_translator::translation_loop() {
         {
             co_await _wait_for_scheduler_cb.wait(
               [this] { return _translation_state.has_value(); });
+            auto clear_finish_request = ss::defer(
+              [this] { _finish_translation_requested = false; });
             auto holder = _translation_state->gate.hold();
             auto deadline = _translation_state->translate_for;
             auto start_time = clock::now();

src/v/datalake/translation/tests/scheduler_fixture.h

@@ -43,6 +43,12 @@ class mock_translator : public translator {
     std::chrono::milliseconds current_lag_ms() const override {
         return std::chrono::milliseconds{0};
     }
+    void set_finish_translation() final {
+        _finish_translation_requested = true;
+    }
+    bool get_finish_translation() final {
+        return _finish_translation_requested;
+    }
 
 private:
     // Mock of a single parquet file writer.
@@ -91,6 +97,7 @@ class mock_translator : public translator {
     ss::timer<clock> _translation_timer;
     ss::condition_variable _wait_for_scheduler_cb;
     clock::time_point _next_checkpoint;
+    bool _finish_translation_requested{false};
 };
 
 // A translator that overshoots deadline and requires explict force flushing