[airflow] branch main updated: Do not seek error file when it is closed (#17187)

[ep...@apache.org]

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

ephraimanierobi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 01a0aca  Do not seek error file when it is closed (#17187)
01a0aca is described below

commit 01a0aca249eeaf71d182bf537b9d04121257ac09
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Sun Jul 25 16:19:18 2021 +0100

    Do not seek error file when it is closed (#17187)
    
    We do not check if error file is closed before we seek it, which causes exceptions.
    Sometimes, this error file does not exist e.g when the task state is changed externally.
    
    This change fixes it by returning None when the file is closed so that custom text can be used for error.
    
    Co-authored-by: Jed Cunningham <66...@users.noreply.github.com>
---
 airflow/models/taskinstance.py    |  2 ++
 tests/models/test_taskinstance.py | 13 +++++++++++++
 2 files changed, 15 insertions(+)

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index a71230f..6fb437b 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -112,6 +112,8 @@ def set_current_context(context: Context):
 
 def load_error_file(fd: IO[bytes]) -> Optional[Union[str, Exception]]:
     """Load and return error from error file"""
+    if fd.closed:
+        return None
     fd.seek(0, os.SEEK_SET)
     data = fd.read()
     if not data:
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index 1656820..39c99da 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -21,6 +21,7 @@ import os
 import time
 import unittest
 import urllib
+from tempfile import NamedTemporaryFile
 from typing import List, Optional, Union, cast
 from unittest import mock
 from unittest.mock import call, mock_open, patch
@@ -48,6 +49,7 @@ from airflow.models import (
     TaskReschedule,
     Variable,
 )
+from airflow.models.taskinstance import load_error_file, set_error_file
 from airflow.operators.bash import BashOperator
 from airflow.operators.dummy import DummyOperator
 from airflow.operators.python import PythonOperator
@@ -119,6 +121,17 @@ class TestTaskInstance(unittest.TestCase):
     def tearDown(self):
         self.clean_db()
 
+    def test_load_error_file_returns_None_for_closed_file(self):
+        error_fd = NamedTemporaryFile()
+        error_fd.close()
+        assert load_error_file(error_fd) is None
+
+    def test_load_error_file_loads_correctly(self):
+        error_message = "some random error message"
+        with NamedTemporaryFile() as error_fd:
+            set_error_file(error_fd.name, error=error_message)
+            assert load_error_file(error_fd) == error_message
+
     def test_set_task_dates(self):
         """
         Test that tasks properly take start/end dates from DAGs