[airflow] 21/45: Added logging to get better diagnosis of flaky backfill test (#25106)

[ep...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ephraimanierobi pushed a commit to branch v2-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 637af6a06d95a617ef2e1eb58e95e92af7f1d2d5
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Sat Jul 16 17:15:48 2022 +0200

    Added logging to get better diagnosis of flaky backfill test (#25106)
    
    (cherry picked from commit 7850dc3c714ef188014a08aae63b4d344242208e)
---
 airflow/models/dagrun.py        | 5 ++++-
 tests/jobs/test_backfill_job.py | 5 ++++-
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index ad0dcdfebd..8589f6878e 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -208,11 +208,14 @@ class DagRun(Base, LoggingMixin):
 
     def __repr__(self):
         return (
-            '<DagRun {dag_id} @ {execution_date}: {run_id}, externally triggered: {external_trigger}>'
+            '<DagRun {dag_id} @ {execution_date}: {run_id}, state:{state}, '
+            'queued_at: {queued_at}. externally triggered: {external_trigger}>'
         ).format(
             dag_id=self.dag_id,
             execution_date=self.execution_date,
             run_id=self.run_id,
+            state=self.state,
+            queued_at=self.queued_at,
             external_trigger=self.external_trigger,
         )
 
diff --git a/tests/jobs/test_backfill_job.py b/tests/jobs/test_backfill_job.py
index 5222f9cdbe..39c215991f 100644
--- a/tests/jobs/test_backfill_job.py
+++ b/tests/jobs/test_backfill_job.py
@@ -987,13 +987,15 @@ class TestBackfillJob:
                         dag_id=dag_id,
                     )
                     dag_maker.create_dagrun(
-                        state=None,
+                        state=State.RUNNING,
                         # Existing dagrun that is not within the backfill range
                         run_id=run_id,
                         execution_date=DEFAULT_DATE + datetime.timedelta(hours=1),
                     )
                     thread_session.commit()
                     cond.notify()
+                except Exception:
+                    logger.exception("Exception when creating DagRun")
                 finally:
                     cond.release()
                     thread_session.close()
@@ -1016,6 +1018,7 @@ class TestBackfillJob:
                 # reached, so it is waiting
                 dag_run_created_cond.wait(timeout=1.5)
                 dagruns = DagRun.find(dag_id=dag_id)
+                logger.info("The dag runs retrieved: %s", dagruns)
                 assert 1 == len(dagruns)
                 dr = dagruns[0]
                 assert dr.run_id == run_id