task_group expand_kwargs with task response list gives - TypeError: Object of type PlainXcomArg is not JSON serializable.

[satyadev0152]

Airflow version: 2.6.3
Google Composer version: 2.5.2

Hi Everyone,

I am trying to create a dag which has a task that returns multiple outputs with one of the return values a list of dictionaries.
I have another task_group containing 3 simplehttpoperator tasks. The idea is to do dynamic task mapping over the task_group using expand_kwargs which will get the list of dictionaries as input from the task.

Please find the rough example dag script below:

@dag()
def sample_dag():
    @task(multiple_outputs=True)
    def list_of_dicts():
        output= {'list_of_dicts': [{input_1: input_value_1, input_2: input_value_2}, {input_1: input_value_1, input_2: input_value_2}]}
        return output

    @task_group
    def task_group(input_1, input_2):
        simplehttpoperator_task1 = SimpleHttpOperator(
            data = json.dumps({x: input_1, y=input_2})
        simplehttpoperator_task1

    list_of_dicts = list_of_dicts()
    task_group.expand_kwargs(list_of_dicts['output'])

The AIrflow UI shows the error
Broken DAG (dags/accelerate_dags/schema_update_dag_2.py): Traceback (most recent call last): File "/opt/python3.8/lib/python3.8/json/encoder.py", line 257, in iterencode return _iterencode(o, 0) File "/opt/python3.8/lib/python3.8/json/encoder.py", line 179, in default raise TypeError(f'Object of type {o._class.name_} ' TypeError: Object of type PlainXComArg is not JSON serializable

I have tried running the dag with the task alone and the return values seem to be fine.
Only when I try to expand the task do I see the issue.
Sorry for not being able to provide proper code or the outputs that it generates since the data cannot be published elsewhere.

I am happy to provide any more info that can help me with this.
Any help is really appreciated.
Thank you in advance.