How do I pass xcom to 3rd party (traditional) operator when using TaskFlow?

[jamiekt]

I am using Amazon's EmrServerlessCreateApplicationOperator operator. I've gotten the basic functionality working like this:

from airflow.providers.amazon.aws.operators.emr import EmrServerlessCreateApplicationOperator
create_app = EmrServerlessCreateApplicationOperator(
                task_id="create_spark_app_on_emr_serverless",
                job_type="SPARK",
                release_label="emr-6.9.0",
                config={...},
                executor_config=executor_config,
                aws_conn_id="",
            )

However that value passed to parameter config is sufficiently complex that I'd like to construct it in upstream tasks and pass it via xcom because:

• I want to visualise as separate tasks so that my DAG visually represents the work going on
• The config value will get exposed explicitly in the XCom tab in the UI
• I've never used TaskFlow API before and this is a good way of getting familiar with it

Here's what I have:

from airflow.providers.amazon.aws.operators.emr import EmrServerlessCreateApplicationOperator

        with TaskGroup(group_id="create_emr_serverless_app") as create_emr_serverless_app_group:
            @task(task_id="network_configuration")
            def network_configuration() -> dict:
                return {
                    "subnetIds": ["subnet-ABC"],
                    "securityGroupIds": ["sg-12345"],
                }
            @task(task_id="maximum_capacity")
            def maximum_capacity() -> dict:
                return {
                    "cpu": "20vCPU",
                    "memory": "64GB",
                }
            @task(task_id="auto_stop_configuration")
            def auto_stop_configuration() -> dict:
                return {
                    "enabled": True,
                    "idleTimeoutMinutes": 15,
                }
            @task(task_id="config")
            def config(network_configuration: dict, maximum_capacity: dict, tags: dict, auto_stop_configuration: dict) -> dict:
                return {
                    "name": "my-emv-serverless-app",
                    "networkConfiguration": network_configuration,
                    "maximumCapacity": maximum_capacity,
                    "autoStopConfiguration": auto_stop_configuration,
                }
            create_slo_app = EmrServerlessCreateApplicationOperator(
                task_id="create_spark_app_on_emr_serverless",
                job_type="SPARK",
                release_label="emr-6.9.0",
                # ***** All upstream tasks being called here *****
                config=config(network_configuration(), maximum_capacity(), auto_stop_configuration()),
                executor_config=executor_config,
                aws_conn_id="",
            )

I would have expected that task config would appear upstream of create_spark_app_on_emr_serverless but that hasn't happened:

Furthermore when I execute the DAG task create_spark_app_on_emr_serverless fails with err:

Traceback (most recent call last):
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/providers/amazon/aws/operators/emr.py", line 535, in execute
response = self.hook.conn.create_application(
TypeError: botocore.client.ClientCreator._create_api_method.._api_call() argument after ** must be a mapping, not PlainXComArg

From that error message I assume that the correct value is not getting passed to create_spark_app_on_emr_serverless. I don't know why that is so.

Please can someone tell me how to pass the value returned from task config to the config parameter of task create_spark_app_on_emr_serverless?

note that I have verified that return_value from task config is as expected