You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "SKalide (via GitHub)" <gi...@apache.org> on 2024/08/14 11:05:42 UTC
[I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
SKalide opened a new issue, #41470:
URL: https://github.com/apache/airflow/issues/41470
### Apache Airflow version
2.9.3
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
BigQueryTableExistenceSensor fire an "execute cannot be called outside TaskInstance" warning on execute
My observation: The ExecutorSafeguard is being called twice for a single sensor execution
Reason is, that the first encounter is over the TaskInstance `_execute_callable` (as intended). Now the decorater checks, if it is executed within a Taskinstance. After that, it will call the "real" `execute` function, which is also decorated and therefor will do the check again, but without a sentinel, which leads to a failed check and fires a warning (or raise a exception if allow_nested_operators is set).
See also: https://github.com/apache/airflow/discussions/41426
And there was also already opened an issue with the same problem but not solved, see [#39413
](https://github.com/apache/airflow/issues/39413)
### What you think should happen instead?
Decorater should not fire a warning. Maybe it should be checked, if the decorater was already called and therefor did already checked it.
### How to reproduce
from airflow import DAG
from airflow.providers.google.cloud.sensors.bigquery import BigQueryTableExistenceSensor
from airflow.utils.dates import days_ago
from airflow.models.baseoperator import BaseOperator
class LoggingBaseOperator(BaseOperator):
def execute(self, context):
self.log.info(f"Executing {self.__class__.__name__}")
return super().execute(context)
class LoggingBigQueryTableExistenceSensor(
BigQueryTableExistenceSensor, LoggingBaseOperator
):
def poke(self, context):
self.log.info(f"Poking {self.__class__.__name__}")
return True
dag = DAG(
"test_bigquery_sensor_double_execution",
default_args={
"start_date": days_ago(1),
},
description="A simple DAG to test BigQueryTableExistenceSensor execution",
schedule_interval=None,
)
sensor_task = LoggingBigQueryTableExistenceSensor(
task_id="test_sensor",
project_id="your-project-id",
dataset_id="your-dataset-id",
table_id="your-table-id",
poke_interval=60,
dag=dag,
)
[log_bg.txt](https://github.com/user-attachments/files/16611919/log_bg.txt)
### Operating System
Ubuntu 22.04.4 LTS
### Versions of Apache Airflow Providers
_No response_
### Deployment
Docker-Compose
### Deployment details
Docker image: apache/airflow:2.9.3-python3.11
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "npatel44 (via GitHub)" <gi...@apache.org>.
npatel44 commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2594146161
This Warning still happens for custom Operator written on top of Provider's operators. (Multiple levels of nesting) . Logic should work for any levels of nesting but I still see this with airflow 2.10.2 and 2.10.4
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "SKalide (via GitHub)" <gi...@apache.org>.
SKalide closed issue #41470: "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance
URL: https://github.com/apache/airflow/issues/41470
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401453034
> I've looked into this issue more deeply and I believe I can provide a solution. I could use some guidance on whether I should submit a PR for this.
>
> The problem, in my opinion, is that after `_execute_callable` is called, it enters the `ExecutorSafeguard` wrapper. There, it checks if it's called inside a `TaskInstance` (hence the `_execute_callable`) and then returns the execution function itself. Because the `execute` method of BigQueryTableExistenceSensor is also decorated, it automatically re-enters the `ExecutorSafeguard` wrapper, but this time it appears as if it was called outside a Task Instance, causing the check to fail. With `allow_nested_operators`, it only fires a warning instead of failing.
>
> My proposed solution is to check if the `ExecutorSafeguard` has already been called in the current execution context. If so, we can skip the check for nested calls. This can be achieved by setting a flag using thread-local storage to ensure thread safety.
>
> ```python
> class ExecutorSafeguard:
> """
> The ExecutorSafeguard decorator.
>
> Checks if the execute method of an operator isn't manually called outside
> the TaskInstance as we want to avoid bad mixing between decorated and
> classic operators.
> """
>
> test_mode = conf.getboolean("core", "unit_test_mode")
> _local = local()
>
> @classmethod
> def decorator(cls, func):
> @wraps(func)
> def wrapper(self, *args, **kwargs):
> if (
> getattr(ExecutorSafeguard._local, "in_executor_safeguard", False)
> and self.allow_nested_operators
> ):
> # If already in ExecutorSafeguard, call execution function - recursive call
> return func(self, *args, **kwargs)
> ExecutorSafeguard._local.in_executor_safeguard = True
>
> try:
> from airflow.decorators.base import DecoratedOperator
>
> sentinel = kwargs.pop(f"{self.__class__.__name__}__sentinel", None)
>
> if (
> not cls.test_mode
> and sentinel != _sentinel
> and not isinstance(self, DecoratedOperator)
> ):
> message = (
> f"{self.__class__.__name__}.{func.__name__} cannot be called outside TaskInstance!"
> )
> raise AirflowException(message)
> return func(self, *args, **kwargs)
> finally:
> ExecutorSafeguard._local.in_executor_safeguard = False
>
> return wrapper
> ```
>
> This solution would suppress the warning for legitimate nested calls while still protecting against unintended external calls. The use of `local()` ensures that this check is thread-safe.
>
> I'm willing to create a Pull Request with this solution if the maintainers think this approach is appropriate. What do you think about this solution? Are there any aspects I should consider or modify before proceeding with a PR?
I like the approach, you can open a PR and add you example as a test case, that way we are sure this case is not only fixed but also tested against regression.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "SKalide (via GitHub)" <gi...@apache.org>.
SKalide commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2302875072
I've looked into this issue more deeply and I believe I can provide a solution. I could use some guidance on whether I should submit a PR for this.
The problem, in my opinion, is that after `_execute_callable` is called, it enters the `ExecutorSafeguard` wrapper. There, it checks if it's called inside a `TaskInstance` (hence the `_execute_callable`) and then returns the execution function itself. Because the `execute` method of BigQueryTableExistenceSensor is also decorated, it automatically re-enters the `ExecutorSafeguard` wrapper, but this time it appears as if it was called outside a Task Instance, causing the check to fail. With `allow_nested_operators`, it only fires a warning instead of failing.
My proposed solution is to check if the `ExecutorSafeguard` has already been called in the current execution context. If so, we can skip the check for nested calls. This can be achieved by setting a flag using thread-local storage to ensure thread safety.
```
class ExecutorSafeguard:
"""
The ExecutorSafeguard decorator.
Checks if the execute method of an operator isn't manually called outside
the TaskInstance as we want to avoid bad mixing between decorated and
classic operators.
"""
test_mode = conf.getboolean("core", "unit_test_mode")
_local = local()
@classmethod
def decorator(cls, func):
@wraps(func)
def wrapper(self, *args, **kwargs):
if (
getattr(ExecutorSafeguard._local, "in_executor_safeguard", False)
and self.allow_nested_operators
):
# If already in ExecutorSafeguard, call execution function - recursive call
return func(self, *args, **kwargs)
ExecutorSafeguard._local.in_executor_safeguard = True
try:
from airflow.decorators.base import DecoratedOperator
sentinel = kwargs.pop(f"{self.__class__.__name__}__sentinel", None)
if (
not cls.test_mode
and sentinel != _sentinel
and not isinstance(self, DecoratedOperator)
):
message = (
f"{self.__class__.__name__}.{func.__name__} cannot be called outside TaskInstance!"
)
raise AirflowException(message)
return func(self, *args, **kwargs)
finally:
ExecutorSafeguard._local.in_executor_safeguard = False
return wrapper
```
This solution would suppress the warning for legitimate nested calls while still protecting against unintended external calls. The use of `local()` ensures that this check is thread-safe.
I'm willing to create a Pull Request with this solution if the maintainers think this approach is appropriate. What do you think about this solution? Are there any aspects I should consider or modify before proceeding with a PR?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "adam-aczel-towa (via GitHub)" <gi...@apache.org>.
adam-aczel-towa commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2633391896
I'm getting the same warning for a ShortCircuitOperator in Version 2.10.4
`{baseoperator.py:421} WARNING - ShortCircuitOperator.execute cannot be called outside TaskInstance!`
The DAG configuration looks like this:
` # Create Airflow Operators for each task dynamically
task_objects = {}
for task_id, python_callable in tasks:
if task_id.startswith('get_'):
task_objects[task_id] = ShortCircuitOperator(
task_id=task_id,
python_callable=python_callable,
op_kwargs={
'credentials': credentials
},
dag=dag,
)
else:
task_objects[task_id] = PythonOperator(
task_id=task_id,
python_callable=python_callable,
op_kwargs={
'credentials': credentials
},
dag=dag,
)`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401785465
The build of the [PR](https://github.com/apache/airflow/pull/42849) passed so once it's merged I think we can close this issue.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2288460254
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401447224
> @SKalide i think you can simplify your example:
>
> ```python
> from airflow import DAG
> from airflow.utils.dates import days_ago
> from airflow.operators.bash import BashOperator
>
>
> class MyBashOperator(BashOperator):
> def execute(self, context):
> self.log.info(f"Executing {self.__class__.__name__}")
> return super().execute(context)
>
>
> dag = DAG(
> "test_bash_execution",
> default_args={
> "start_date": days_ago(1),
> },
> description="A simple DAG to test MyBashOperator execution",
> schedule_interval=None,
> )
>
> task = MyBashOperator(
> task_id="my_bash_operator",
> bash_command="ls -lah",
> dag=dag,
> )
> ```
>
> this produces the same warning message by using this simpler example, i believe it's more convincing that this issue is "serious" 😁
I think I understand the issue of the warning and I think some refactoring will be needed to avoid the warning in the case you extend an existing operator, as that should still be allowed. Will have to create a test case for it and then check how we can prevent the warning in that case. The warning is there to prevent people from calling other operators from python operators, but in you example its not the case as it's inheritance.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2450121982
> Hi @dabla, I'm using AWS MWAA 2.10.1 and I still get the same warning. Just wondering whether the issue is fixed in 2.10.1 or only in 2.9.3
>
> Script,
>
> ```
> from airflow import DAG
> from airflow.utils.dates import days_ago
> from airflow.operators.bash import BashOperator
> from airflow.utils.log.logging_mixin import LoggingMixin
>
> class MyBashOperator(BashOperator, LoggingMixin):
> def execute(self, context):
> self.log.info("Executing MyBashOperator directly in the DAG definition")
> return super().execute(context)
>
> dag = DAG(
> "test_bash_execution",
> default_args={
> "start_date": days_ago(1),
> },
> schedule_interval=None,
> )
>
> # Instantiate the operator
> my_bash_task = MyBashOperator(
> task_id="my_bash_operator",
> bash_command="echo 'Testing direct execution'",
> dag=dag,
> )
> ```
>
> Logs, <img alt="image" width="816" src="https://private-user-images.githubusercontent.com/182846498/381787558-a4fc4980-6751-4a85-97a1-e64fd0a4086c.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MzAzODczMzgsIm5iZiI6MTczMDM4NzAzOCwicGF0aCI6Ii8xODI4NDY0OTgvMzgxNzg3NTU4LWE0ZmM0OTgwLTY3NTEtNGE4NS05N2ExLWU2NGZkMGE0MDg2Yy5wbmc_WC1BbXotQWxnb3JpdGhtPUFXUzQtSE1BQy1TSEEyNTYmWC1BbXotQ3JlZGVudGlhbD1BS0lBVkNPRFlMU0E1M1BRSzRaQSUyRjIwMjQxMDMxJTJGdXMtZWFzdC0xJTJGczMlMkZhd3M0X3JlcXVlc3QmWC1BbXotRGF0ZT0yMDI0MTAzMVQxNTAzNThaJlgtQW16LUV4cGlyZXM9MzAwJlgtQW16LVNpZ25hdHVyZT03ZjQwNjJhNDRiNDIwMTBjMDVkZjBlNWI4Zjc1NThlYTMwNmYwZjRkNjk1MWRmOTQ1YTNlODI3NjhjNDQ3Nzg4JlgtQW16LVNpZ25lZEhlYWRlcnM9aG9zdCJ9._acIyB7ewJl_8Qjfgw8-_mMk0LDX5J5ISGisI7sFsTw">
The fix is not in Airflow 2.10.1 nor 2.10.2, I just checked the source code. I suppose it will be in 2.10.3 @potiuk @uranusjr ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401696393
I've created a [PR](https://github.com/apache/airflow/pull/42849) which should fix this issue.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "SKalide (via GitHub)" <gi...@apache.org>.
SKalide closed issue #41470: "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance
URL: https://github.com/apache/airflow/issues/41470
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2459272506
> Hi @dabla, I'm using AWS MWAA 2.10.1 and I still get the same warning. Just wondering whether the issue is fixed in 2.10.1 or only in 2.9.3
>
> Script,
>
> ```
> from airflow import DAG
> from airflow.utils.dates import days_ago
> from airflow.operators.bash import BashOperator
> from airflow.utils.log.logging_mixin import LoggingMixin
>
> class MyBashOperator(BashOperator, LoggingMixin):
> def execute(self, context):
> self.log.info("Executing MyBashOperator directly in the DAG definition")
> return super().execute(context)
>
> dag = DAG(
> "test_bash_execution",
> default_args={
> "start_date": days_ago(1),
> },
> schedule_interval=None,
> )
>
> # Instantiate the operator
> my_bash_task = MyBashOperator(
> task_id="my_bash_operator",
> bash_command="echo 'Testing direct execution'",
> dag=dag,
> )
> ```
>
> Logs, <img alt="image" width="816" src="https://private-user-images.githubusercontent.com/182846498/381787558-a4fc4980-6751-4a85-97a1-e64fd0a4086c.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MzA4ODkxMTAsIm5iZiI6MTczMDg4ODgxMCwicGF0aCI6Ii8xODI4NDY0OTgvMzgxNzg3NTU4LWE0ZmM0OTgwLTY3NTEtNGE4NS05N2ExLWU2NGZkMGE0MDg2Yy5wbmc_WC1BbXotQWxnb3JpdGhtPUFXUzQtSE1BQy1TSEEyNTYmWC1BbXotQ3JlZGVudGlhbD1BS0lBVkNPRFlMU0E1M1BRSzRaQSUyRjIwMjQxMTA2JTJGdXMtZWFzdC0xJTJGczMlMkZhd3M0X3JlcXVlc3QmWC1BbXotRGF0ZT0yMDI0MTEwNlQxMDI2NTBaJlgtQW16LUV4cGlyZXM9MzAwJlgtQW16LVNpZ25hdHVyZT00YzQ3OGE1MzJlZTkyYjBmNWY5N2M0MTk2MWJmZGY5MjNmMzk1NTdkNGE5MGU1MDQxZmM5YjM4OThiYmViYWNmJlgtQW16LVNpZ25lZEhlYWRlcnM9aG9zdCJ9.69u-HvJ6oqDJtfQ67BrCIIoFTuOfG6QfceXufwpB-6o">
@sam-gen-cop The fix is now available in Airflow 2.10.3
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "sam-gen-cop (via GitHub)" <gi...@apache.org>.
sam-gen-cop commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2448808728
Hi, I'm using AWS MWAA 2.10.1 and I still get the same warning. Just wondering whether the issue is fixed in 2.10.1 or only in 2.9.3
Script,
```
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash import BashOperator
from airflow.utils.log.logging_mixin import LoggingMixin
class MyBashOperator(BashOperator, LoggingMixin):
def execute(self, context):
self.log.info("Executing MyBashOperator directly in the DAG definition")
return super().execute(context)
dag = DAG(
"test_bash_execution",
default_args={
"start_date": days_ago(1),
},
description="A simple DAG to test MyBashOperator execution directly",
schedule_interval=None,
)
# Instantiate the operator
my_bash_task = MyBashOperator(
task_id="my_bash_operator",
bash_command="echo 'Testing direct execution'",
dag=dag,
)
```
Logs,
<img width="816" alt="image" src="https://github.com/user-attachments/assets/a4fc4980-6751-4a85-97a1-e64fd0a4086c">
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "HunterAnnan (via GitHub)" <gi...@apache.org>.
HunterAnnan commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2948936792
I'm also getting this in Version 2.10.5 for `GCSObjectExistenceSensor`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2596751923
> This Warning still happens for custom Operator written on top of Provider's operators. (Multiple levels of nesting) . Logic of this PR should work for any levels of nesting so not sure what's the issue but I still see this with airflow 2.10.2 and 2.10.4
Can you open a new issue for that with the detail of your case please ? Or even better - contributing a fix - following that one here should be relatively simple so maybe you or your company would like to contribute it? Just to put it in perspective - this is an open-source project that a lot of people contribute to, so when you create such an issue, you will have to wait until somoene picks it up and fix it. But the most certain way is to implement and contribute a fix yourself. The second best is to find someone who can do it - maybe even paying them. The third best is to open a detailed issue that makes it easy to reproduce by anyone, we can then mark it as "good first issue" for example and hopefully sooner or later one of the contributors will fix it.
Following one of those routes is the best way to help in solving your issue.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2596769590
I’m wondering if there wasn’t even an additional fix for this issue but only made it in main/Airflow 3 codebase but could be mistalen ofc.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401543031
> ```python
> class LoggingBaseOperator(BaseOperator):
> def execute(self, context):
> self.log.info(f"Executing {self.__class__.__name__}")
> return super().execute(context)
> ```
I've been able to reproduce the issue in a test case, will open a PR for it which will fix it, will try the solution proposed by @SKalide
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2401682183
> I've looked into this issue more deeply and I believe I can provide a solution. I could use some guidance on whether I should submit a PR for this.
>
> The problem, in my opinion, is that after `_execute_callable` is called, it enters the `ExecutorSafeguard` wrapper. There, it checks if it's called inside a `TaskInstance` (hence the `_execute_callable`) and then returns the execution function itself. Because the `execute` method of BigQueryTableExistenceSensor is also decorated, it automatically re-enters the `ExecutorSafeguard` wrapper, but this time it appears as if it was called outside a Task Instance, causing the check to fail. With `allow_nested_operators`, it only fires a warning instead of failing.
>
> My proposed solution is to check if the `ExecutorSafeguard` has already been called in the current execution context. If so, we can skip the check for nested calls. This can be achieved by setting a flag using thread-local storage to ensure thread safety.
>
> ```python
> class ExecutorSafeguard:
> """
> The ExecutorSafeguard decorator.
>
> Checks if the execute method of an operator isn't manually called outside
> the TaskInstance as we want to avoid bad mixing between decorated and
> classic operators.
> """
>
> test_mode = conf.getboolean("core", "unit_test_mode")
> _local = local()
>
> @classmethod
> def decorator(cls, func):
> @wraps(func)
> def wrapper(self, *args, **kwargs):
> if (
> getattr(ExecutorSafeguard._local, "in_executor_safeguard", False)
> and self.allow_nested_operators
> ):
> # If already in ExecutorSafeguard, call execution function - recursive call
> return func(self, *args, **kwargs)
> ExecutorSafeguard._local.in_executor_safeguard = True
>
> try:
> from airflow.decorators.base import DecoratedOperator
>
> sentinel = kwargs.pop(f"{self.__class__.__name__}__sentinel", None)
>
> if (
> not cls.test_mode
> and sentinel != _sentinel
> and not isinstance(self, DecoratedOperator)
> ):
> message = (
> f"{self.__class__.__name__}.{func.__name__} cannot be called outside TaskInstance!"
> )
> raise AirflowException(message)
> return func(self, *args, **kwargs)
> finally:
> ExecutorSafeguard._local.in_executor_safeguard = False
>
> return wrapper
> ```
>
> This solution would suppress the warning for legitimate nested calls while still protecting against unintended external calls. The use of `local()` ensures that this check is thread-safe.
>
> I'm willing to create a Pull Request with this solution if the maintainers think this approach is appropriate. What do you think about this solution? Are there any aspects I should consider or modify before proceeding with a PR?
The above proposed solution doesn't work for all test cases, especially in case of mixed usage with task decorated function calling execute methods on native operator. I've inspired the final working solution from it and now it should also support the case you presented.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] "execute cannot be called outside TaskInstance" warning fired despite being in a TaskInstance [airflow]
Posted by "zachliu (via GitHub)" <gi...@apache.org>.
zachliu commented on issue #41470:
URL: https://github.com/apache/airflow/issues/41470#issuecomment-2400722201
@SKalide i think you can simplify your example:
```python
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash import BashOperator
class MyBashOperator(BashOperator):
def execute(self, context):
self.log.info(f"Executing {self.__class__.__name__}")
return super().execute(context)
dag = DAG(
"test_bash_execution",
default_args={
"start_date": days_ago(1),
},
description="A simple DAG to test MyBashOperator execution",
schedule_interval=None,
)
task = MyBashOperator(
task_id="my_bash_operator",
bash_command="ls -lah",
dag=dag,
)
```
this produces the same warning message
by using this simpler example, i believe it's more convincing that this issue is "serious" :grin:
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org