BranchingOperator fails when using custom backend xcom deserialization with NoCredentialsError
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
2.10.5
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
I'm encountering an issue when deserializing XComs using a custom XCom backend that loads data from S3 using `S3Hook(aws_conn_id='aws')`. `aws` is the connection Id defined with the aws credentials. The same code is working for Airflow 2.10.4 but not anymore from 2.10.5 and further.
Despite having a valid AWS connection configured in Airflow (visible and **working well** in **ALL** other parts of the DAG), I get the following error during deserialization steps:
```botocore.exceptions.NoCredentialsError: Unable to locate credentials````
With the function:
```
hook = S3Hook(aws_conn_id='aws')
head_object = hook.head_object(key=key, bucket_name='bucket')
```
This error is not visible in the Airflow UI but I found it in the scheduler container's logs. The branching task is set as success but the following branched task is queued 'forever'. I assume this error is coming from the following task which is pulling a xcom.
After switching to manual session initialization using `boto3.Session(...)`, I still have the same error... BUT it is working and I am settings the aws credentials directly inside the code, without trying to get the `aws` Connection Id. This is telling me that in the specific case of BranchingOperator, something is happening and the follwing task' context doesn't have the connection Id anymore.
### What you think should happen instead?
The 'aws' connection is not resolved correctly anymore by the scheduler, even though it is defined in Airflow. The context might have changed?
### How to reproduce
```
class CustomXComBackendPandas(BaseXCom):
XCOM_PREFIX = "xcom_s3://"
S3_PATH_PREFIX = "tmp"
_BUCKET_NAME = ""
DATAFRAME_EXTENSION = ".parquet"
@staticmethod
def deserialize_value(result: BaseXCom) -> Any:
hook = S3Hook(aws_conn_id='aws')
head_object = hook.head_object(key='key', bucket_name='bucket')
```
### Operating System
macOS 15.4.1 (Sonoma)
### Versions of Apache Airflow Providers
_No response_
### Deployment
Astronomer
### Deployment details
_No response_
### 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)
Contributor guide
Research direction
Start by reproducing the failure through CustomXComBackendPandas.deserialize_value, focusing on S3Hook(aws_conn_id='aws'). Trace the BranchingOperator and scheduler path that performs deserialization, comparing Airflow 2.10.4 with 2.10.5. Done means the configured aws connection is resolved during deserialization and the branched downstream task is not left queued.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, python
- Domain
- backend, cloud
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Needs clarification
- Newbie friendliness
- 45/100