turbaszek commented on a change in pull request #7542:
URL: https://github.com/apache/airflow/pull/7542#discussion_r420705597
##########
File path: airflow/executors/celery_executor.py
##########
@@ -319,3 +262,109 @@ def execute_async(self,
def terminate(self):
pass
+
+
+def fetch_celery_task_state(async_result: AsyncResult) -> Tuple[str,
Union[str, ExceptionWithTraceback]]:
+ """
+ Fetch and return the state of the given celery task. The scope of this
function is
+ global so that it can be called by subprocesses in the pool.
+
+ :param async_result: a tuple of the Celery task key and the async Celery
object used
+ to fetch the task's state
+ :type async_result: tuple(str, celery.result.AsyncResult)
+ :return: a tuple of the Celery task key and the Celery state of the task
+ :rtype: tuple[str, str]
+ """
+
+ try:
+ with timeout(seconds=OPERATION_TIMEOUT):
+ # Accessing state property of celery task will make actual network
request
+ # to get the current state of the task
+ return async_result.task_id, async_result.state
+ except Exception as e: # pylint: disable=broad-except
+ exception_traceback = "Celery Task ID: {}\n{}".format(async_result,
traceback.format_exc())
Review comment:
```suggestion
exception_traceback = f"Celery Task ID:
{async_result}\n{traceback.format_exc()}"
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
