From 65113c3c5fff9d3947c7fda887024ed24c49e6b5 Mon Sep 17 00:00:00 2001
From: jamiedemaria <demaria.jamie@gmail.com>
Date: Tue, 10 Dec 2024 14:48:55 -0500
Subject: [PATCH] change backfill log message when waiting for runs to retry
 (#26360)

## Summary & Motivation
Updates the log message when a backfill is waiting for runs to retry to
log the run ids it is waiting on.

## How I Tested These Changes
forced a backfill into a situation where these logs get printed. output
```
2024-12-09 15:40:44 -0500 - dagster.daemon.BackfillDaemon - INFO - The following runs for the backfill will be retried, but have not been launched. Backfill is still in progress:
1908f288-6c03-4620-92ec-36b8fb24888e
af189906-96ba-467d-9ea0-d74be361a880
9a98a638-2ab5-4cd0-8d27-02cf40352adc
```
---
 .../dagster/_core/execution/asset_backfill.py | 27 +++++++++++--------
 1 file changed, 16 insertions(+), 11 deletions(-)

diff --git a/python_modules/dagster/dagster/_core/execution/asset_backfill.py b/python_modules/dagster/dagster/_core/execution/asset_backfill.py
index 82eedb6e5eb04..590ad6f7ecb82 100644
--- a/python_modules/dagster/dagster/_core/execution/asset_backfill.py
+++ b/python_modules/dagster/dagster/_core/execution/asset_backfill.py
@@ -57,6 +57,7 @@
     ASSET_PARTITION_RANGE_START_TAG,
     BACKFILL_ID_TAG,
     PARTITION_NAME_TAG,
+    WILL_RETRY_TAG,
 )
 from dagster._core.utils import make_new_run_id, toposort
 from dagster._core.workspace.context import BaseWorkspaceRequestContext, IWorkspaceProcessContext
@@ -964,19 +965,23 @@ def backfill_is_complete(
         logger.info("Backfill has in progress runs. Backfill is still in progress.")
         return False
     # Condition 3 - if there are runs that will be retried, but have not yet been retried, the backfill is not complete
-    if any(
-        [
-            run.is_complete_and_waiting_to_retry
-            for run in instance.get_runs(
-                filters=RunsFilter(
-                    tags={BACKFILL_ID_TAG: backfill_id},
-                    statuses=[DagsterRunStatus.FAILURE],
-                )
+    runs_waiting_to_retry = [
+        run.run_id
+        for run in instance.get_runs(
+            filters=RunsFilter(
+                tags={BACKFILL_ID_TAG: backfill_id, WILL_RETRY_TAG: "true"},
+                statuses=[DagsterRunStatus.FAILURE],
             )
-        ]
-    ):
+        )
+        if run.is_complete_and_waiting_to_retry
+    ]
+    if len(runs_waiting_to_retry) > 0:
+        num_runs_to_log = 20
+        formatted_runs = "\n".join(runs_waiting_to_retry[:num_runs_to_log])
+        if len(runs_waiting_to_retry) > num_runs_to_log:
+            formatted_runs += f"\n... {len(runs_waiting_to_retry) - num_runs_to_log} more"
         logger.info(
-            "Some runs for the backfill will be retried, but have not been launched. Backfill is still in progress."
+            f"The following runs for the backfill will be retried, but retries have not been launched. Backfill is still in progress:\n{formatted_runs}"
         )
         return False
     return True