dl/translation: fix busy loop under coordinator backpressure

The translation loop treated a backpressured fetch as a successful
iteration and canceled its retry jitter, so backpressured translators
re-polled the coordinator as fast as the fetch RPC completed. This added
noticeable CPU churn in some clusters with existing backpressure.

This commit routes backpressured iterations through the same jittered
sleep as a failed fetch; a requested finish still proceeds.

Author: andrwng

src/v/datalake/translation/partition_translator.cc

@@ -236,6 +236,7 @@ partition_translator::fetch_translation_offsets(retry_chain_node& rcn) {
           "[{}] Coordinator applying backpressure (too many pending files); "
           "backing off translation",
           _data_source->ntp());
+        offsets.backpressure = true;
         co_return offsets;
     }
 
@@ -474,7 +475,15 @@ ss::future<> partition_translator::translate_until_stopped() {
         if (finish_now) {
             vlog(_logger.debug, "Requested for immediate finish");
         }
-        if (!offsets && !finish_now) {
+        if (!offsets) {
+            // Without reconciled offsets there is nothing to translate or
+            // finish against.
+            continue;
+        }
+        // A backpressured iteration takes the jittered retry path like a
+        // failed fetch, rather than immediately polling the coordinator
+        // again. A requested finish still proceeds below.
+        if (offsets->backpressure && !finish_now) {
             continue;
         }
         if (offsets->next_translation_begin_offset && !finish_now) {

src/v/datalake/translation/partition_translator.h

@@ -163,6 +163,9 @@ class partition_translator : public scheduling::translator {
         // Offset to begin the next translation from.
         // set if there is new data to translate
         std::optional<kafka::offset> next_translation_begin_offset;
+        // Set if the coordinator is shedding load; the translation loop
+        // should back off before polling again.
+        bool backpressure{false};
     };
     ss::future<std::optional<translation_offsets>>
     fetch_translation_offsets(retry_chain_node&);

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

@@ -76,6 +76,8 @@ class fake_test_ctx {
 
     void note_backpressure_rejection() { _backpressure_rejections++; }
 
+    size_t backpressure_rejections() const { return _backpressure_rejections; }
+
     ss::future<> wait_for_backpressure_rejections(
       size_t n, std::chrono::seconds timeout = 10s) {
         RPTEST_REQUIRE_EVENTUALLY_CORO(
@@ -509,6 +511,25 @@ TEST_F_CORO(partition_translator_fixture, test_coordinator_backpressure) {
     ASSERT_GT_CORO(test_ctx.max_translated_offset(), kafka::offset{0});
 }
 
+TEST_F_CORO(
+  partition_translator_fixture, test_coordinator_backpressure_is_paced) {
+    auto& test_ctx = make_test_context();
+    test_ctx.set_coordinator_backpressure(true);
+    test_ctx.set_should_finish_inflight_translation(false);
+    co_await add_translator(test_ctx);
+
+    co_await test_ctx.wait_for_backpressure_rejections(1);
+    auto baseline = test_ctx.backpressure_rejections();
+    co_await ss::sleep(200ms);
+    auto polls = test_ctx.backpressure_rejections() - baseline;
+
+    // Each backpressured iteration should sleep the loop jitter (10ms base in
+    // this fixture) before polling the coordinator again, so expect ~20 polls.
+    // Regression check for a bug where we would poll without waiting: a
+    // translator that spins on fetches would rack up thousands.
+    ASSERT_LE_CORO(polls, 100);
+}
+
 TEST_F_CORO(partition_translator_fixture, test_batching) {
     auto& test_ctx = make_test_context();
     // Let the translator batch to the limit

tests/rptest/tests/datalake/coordinator_backpressure_test.py

@@ -133,7 +133,7 @@ def test_backpressure_on_pending_backlog(self, cloud_storage_type):
                 err_msg="coordinator never applied backpressure",
             )
 
-            # Translators respect the signal by backing off rather than spinning.
+            # Translators respect the signal by backing off.
             wait_until(
                 lambda: self.metric_sum(TRANSLATION_BACKOFF_METRIC) > 0,
                 timeout_sec=30,
@@ -186,6 +186,18 @@ def quiesced():
                 err_msg="pending/translated file counts never stabilized under backpressure",
             )
 
+            # Regression check for a case where translation would spin and
+            # churn despite backpressure: backing off means each translator
+            # sleeps its loop jitter between coordinator polls, so the
+            # cumulative backoff count stays on the order of hundreds, with so
+            # few partitions.
+            backoffs = self.metric_sum(TRANSLATION_BACKOFF_METRIC)
+            self.logger.info(f"translator backoffs while backpressured: {backoffs}")
+            assert backoffs < 10000, (
+                f"translators spun on the backpressured coordinator: "
+                f"{backoffs} backoff loop iterations"
+            )
+
             # Relieve the pressure so the backlog drains promptly.
             self.redpanda.set_cluster_config(
                 {