Serialization problem in trigger in Google provider that raises NotImplementedError in case when non-primitive type is returned from triggerer

[VladaZakharova]

Apache Airflow version

3.0.5

If "Other Airflow 2 version" selected, which one?

No response

What happened?

In Airflow 3.0.3 with PR #51699 we have introduced new logic to serialize objects, that caused multiple failures in Google provider triggers.
There are for now 7 triggers that fail because of "too complicated" objects returned from triggerer on the step when it is trying to serialize the object before passing it to worker. The one example I can give is "airflow.providers.google.cloud.triggers.dataplex.DataplexDataProfileJobTrigger" and the error that is returned from the trigger is:

triggerer  | TriggerRunner().run()
triggerer  | File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 845, in run
triggerer  | asyncio.run(self.arun())
triggerer  | File "/usr/local/lib/python3.10/asyncio/runners.py", line 44, in run
triggerer  | return loop.run_until_complete(main)
triggerer  | File "/usr/local/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
triggerer  | return future.result()
triggerer  | File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 869, in arun
triggerer  | await self.sync_state_to_supervisor(finished_ids)
triggerer  | File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 1055, in sync_state_to_supervisor
triggerer  | resp = await self.comms_decoder.asend(msg)
triggerer  | File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 790, in asend
triggerer  | bytes = frame.as_bytes()
triggerer  | File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/comms.py", line 148, in as_bytes
triggerer  | self.req_encoder.encode_into(self, buffer, 4)
triggerer  | File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/comms.py", line 125, in _msgpack_enc_hook
triggerer  | raise NotImplementedError(f"Objects of type {type(obj)} are not supported")
triggerer  | NotImplementedError: Objects of type <enum 'State'> are not supported
triggerer  | 2025-08-21 12:39:09 [warning  ] Process exited abnormally      [airflow.jobs.triggerer_job_runner] exit_code=126
triggerer  | 2025-08-21 12:39:09 [error    ] Trigger runner process has died! Exiting. [airflow.jobs.triggerer_job_runner]
triggerer  | [2025-08-21T12:39:09.349+0000] {triggerer_job_runner.py:177} INFO - Waiting for triggers to clean up
triggerer  | [2025-08-21T12:39:09.350+0000] {triggerer_job_runner.py:182} INFO - Exited trigger loop

After that the trigger is dying and not doing anything. the task fails with trigger timeout
Not sure if we have issue for that, but the whole list looks like this:

- dataplex_data_profile: kills the triggerer (NotImplementedError: Objects of type <enum 'State'> are not supported)
- dataplex_data_quality: kills the triggerer (NotImplementedError: Objects of type <enum 'State'> are not supported)
- dataproc_batch_deferrable: kills the triggerer (NotImplementedError: Objects of type <enum 'State'> are not supported)
- dataproc_spark_deferrable: kills the triggerer (NotImplementedError: Objects of type <enum 'State'> are not supported)
- example_gcp_transfer: kill the triggerer (NotImplementedError: Objects of type <class 'google.cloud.sotrage_transfer_v1.types.transfer_types.TransferOperation'> are not supported)
- example_transfer_gcs_to_gcs: kill the triggerer (NotImplementedError: Objects of type <class 'google.cloud.sotrage_transfer_v1.types.transfer_types.TransferOperation'> are not supported)
- example_gcp_transfer_aws: kill the triggerer (NotImplementedError: Objects of type <class 'google.cloud.sotrage_transfer_v1.types.transfer_types.TransferOperation'> are not supported)

For all those system tests we observe the same error and I think this should be covered in here to extend the logic for types (introduced in #51699 ):

def _msgpack_enc_hook(obj: Any) -> Any:
    import pendulum

    if isinstance(obj, pendulum.DateTime):
        # convert the pendulm Datetime subclass into a raw datetime so that msgspec can use it's native
        # encoding
        return datetime(
            obj.year, obj.month, obj.day, obj.hour, obj.minute, obj.second, obj.microsecond, tzinfo=obj.tzinfo
        )
    if isinstance(obj, Path):
        return str(obj)
    if isinstance(obj, BaseModel):
        return obj.model_dump(exclude_unset=True)

    # Raise a NotImplementedError for other types
    raise NotImplementedError(f"Objects of type {type(obj)} are not supported")

What you think should happen instead?

No response

How to reproduce

1. Run example_dataplex_dp.py in breeze
2. wait until the task run_data_scan_def will start
3. observe breeze cli to see the actual error in the triggerer. Should look like this:
   triggerer | NotImplementedError: Objects of type <enum 'State'> are not supported
4. the task will die after some time with triggerer timeout

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

This is the problem from airflow core, was introduced in version 3.0.3, starting from this PR: #51699

Deployment

Official Apache Airflow Helm Chart

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

• I agree to follow this project's Code of Conduct

[potiuk]

I think we have two ways of solving it:

1. More generic -> possibly we could serialize those classes (they are protobufs really) using generic to/from String and store the "class" of the object (so that we can deserialize it and know what to deserialize it from)

2. More "extendable" - add "serialization" extension to providers.

We could implement extendable serialization - and have providers to implement "serialize/deserialize" extention. We could have that Google-provider serialzation "extension" recognize and serialize/deserialize the objects we want.

I am torn between the two. 1) is more generic but 2) is something I would also like to see for other providers possibly, and we could have a mechanism that finally allows to add serialization capabilities (and code) independently from airflow-core.

WDYT. @kaxil @amoghrajesh @bolkedebruin @ashb ?

[ashb]

Could we make do with a generic fallback for str(value) in the msg enc hook for anything we don't know about? Would that work here? (It was about to get stored in the DB anyway, so what process did it use to get there anyway?)

[dshvedchenko]

but this is communication channel between python processes, so other part have to deserialize it , so from str it can be tricky

[amoghrajesh]

2. is a much much longer term goal I see and many provider owners would have to kind of chime in to help with that effort and considering 3.1 priorities, now doesn't seem to be a great time for that.

For now, for any object that isn't / cannot be handled by: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/comms.py#L110-L129, maybe we could just stringify it?

[amoghrajesh]

but this is communication channel between python processes, so other part have to deserialize it , so from str it can be tricky

Correct, but that should not be too tricky imo. This is where deserialization is defined:
https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/comms.py#L183-L185. Maybe do something using ast library to get actual object imo.

[stantaneous]

Hi,

We are also facing the similar issue in the standard filesystem provider. We are getting this error in FileSensor - NotImplementedError: Objects of type <class 'pydantic_core._pydantic_core.SerializationIterator'> are not supported.

Is there a patch or quick fix for this issue?

[potiuk]

Hi,

We are also facing the similar issue in the standard filesystem provider. We are getting this error in FileSensor - NotImplementedError: Objects of type <class 'pydantic_core._pydantic_core.SerializationIterator'> are not supported.

Is there a patch or quick fix for this issue?

This is a different issue. 3.0.5 has been yanked due to this issue - you should downgrade to 3.0.4 or wait for 3.0.6 - you can test rc1 that is just being released #54835 -> please test and report in the issue if your problem is fixed @stantaneous

[stantaneous]

Hi,
We are also facing the similar issue in the standard filesystem provider. We are getting this error in FileSensor - NotImplementedError: Objects of type <class 'pydantic_core._pydantic_core.SerializationIterator'> are not supported.
Is there a patch or quick fix for this issue?

This is a different issue. 3.0.5 has been yanked due to this issue - you should downgrade to 3.0.4 or wait for 3.0.6 - you can test rc1 that is just being released #54835 -> please test and report in the issue if your problem is fixed @stantaneous

With 3.0.4, we have the issue with paramiko library as DSSKEY is removed in the paramiko 4.0.0. I was able to fix it by manually adding paramiko<4.0.0 in my requirements.

With 3.0.6rc1, Everything seems to be working fine. Waiting for 3.0.6 :)

[amoghrajesh]

@stantaneous 3.0.4 already had fix for paramiko: #54078 though