Airflow triggerer would fail to complete the trigger job for asset-scheduled triggers due to asset trigger serialization/deserialization issue.

[dshk0718]

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.2.1

What happened and how to reproduce it?

Issue Description
After upgrading to Airflow version 3.2.1 from 3.1.8, the Kafka Message triggerer will always fail to trigger the asset-scheduled DAGs with the following error:
apply_function() got an unexpected keyword argument 'Encoding.VAR'

This happens whether or not other arguments for the apply_function for the KafkaMessageQueueTrigger (which seems to be just a wrapper class that constructs the base MessageQueueTrigger class with the scheme set to kafka) were set using either the apply_function_args or apply_function_kwargs argument.

Based on a deeper investigation into this issue, I've found out that the triggers are serialized differently from how the trigger kwargs values are serialized in version 3.1.8, which currently looks like this version 3.2.1 (some contents intentionally obfuscated due to data being related to our company stuff):

{
  "topics": [
    "some.kafka.topic"
  ],
  "apply_function": "utilities.kafka.apply_function",
  "apply_function_args": {
    "Encoding.VAR": [],
    "Encoding.TYPE": "tuple"
  },
  "apply_function_kwargs": {
    "Encoding.VAR": {
      "action_filter": [
        "create"
      ],
      "data_filter": {
        "Encoding.VAR": {
          "BatchId": null
        },
        "Encoding.TYPE": "dict"
      }
    },
    "Encoding.TYPE": "dict"
  },
  "kafka_config_id": "my_kafka_config",
  "poll_timeout": 1,
  "poll_interval": 1,
  "commit_offset": true
}

In version 3.1.8, the same trigger (with no code change) is serialized as such, which successfully gets decoded and deserialized before being used to successfully trigger the DAG:

{
  "__var": {
    "topics": [
      "some.kafka.topic"
    ],
    "apply_function": "utilities.kafka.apply_function",
    "apply_function_args": {
      "__var": {
        "__var": [],
        "__type": "tuple"
      },
      "__type": "dict"
    },
    "apply_function_kwargs": {
      "__var": {
        "__var": {
          "__var": {
            "action_filter": [
              "create"
            ],
            "data_filter": {
              "__var": {
                "__var": {
                  "__var": {
                    "BatchId": null
                  },
                  "__type": "dict"
                },
                "__type": "dict"
              },
              "__type": "dict"
            }
          },
          "__type": "dict"
        },
        "__type": "dict"
      },
      "__type": "dict"
    },
    "kafka_config_id": "my_kafka_config",
    "poll_timeout": 1,
    "poll_interval": 1
  },
  "__type": "dict"
}

Note that the commit_offset parameter in the kwargs for the apply_function() comes from the update made on the apache-airflow-providers-apache-kafka module version 1.13.0 update.

Due to this difference in the serialized kwargs value for the trigger, the trigger runs to trigger the asset-scheduled DAG will always fail with the aforementioned error saying: apply_function() got an unexpected keyword argument 'Encoding.VAR'.

Steps to Reproduce

1. Set up the triggers, assets (with asset watchers), and DAG; does not need any tasks since the DAG run initialization already fails due to this trigger issue, never executing the DAG code ever; also, whether or not the custom apply_function_kwargs argument is provided should not matter, since the bug is reproducible with or without it:

from airflow.providers.apache.kafka.triggers.msg_queue import KafkaMessageQueueTrigger
from airflow.sdk import Asset, AssetWatcher, dag

TOPICS = ["some.kafka.topic"]

create_trigger = KafkaMessageQueueTrigger(
    topics=TOPICS,
    apply_function="utilities.kafka.apply_function",
    apply_function_kwargs={
        "action_filter": ["create"],
        "data_filter": {"BatchId": None},
    },
    kafka_config_id="my_kafka_config",
    poll_timeout=1,
    poll_interval=1,
)

asset = Asset(
    "kafka_asset_some_entity",
    watchers=[
        AssetWatcher(name="kafka_watcher_create_some_entity", trigger=create_trigger),
    ],
)

@dag(
    schedule=[asset],
    render_template_as_native_obj=True,
)
def create_some_entity():
    pass

create_some_entity()

2. Produce the Kafka message for the given topic, specified for the topics argument in the trigger definition.
3. Wait for the trigger runner to execute the trigger and encounter the error:

[2026-04-27T18:42:54.831211Z] {triggerer_job_runner.py:796} ERROR - Trigger ID 1268 exited with error apply_function() got an unexpected keyword argument 'Encoding.VAR' error_detail=[{'exc_type': 'TypeError', 'exc_value': "apply_function() got an unexpected keyword argument 'Encoding.VAR'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/home/airflow/.local/lib/python3.14/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1071, 'name': 'cleanup_finished_triggers'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1218, 'name': 'run_trigger'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/apache/kafka/triggers/await_message.py', 'lineno': 131, 'name': 'run'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/asgiref/sync.py', 'lineno': 506, 'name': '__call__'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/greenback/_impl.py', 'lineno': 217, 'name': '_greenback_shim'}, {'filename': '/usr/python/lib/python3.14/concurrent/futures/thread.py', 'lineno': 86, 'name': 'run'}, {'filename': '/usr/python/lib/python3.14/concurrent/futures/thread.py', 'lineno': 73, 'name': 'run'}, {'filename': '/home/airflow/.local/lib/python3.14/site-packages/asgiref/sync.py', 'lineno': 561, 'name': 'thread_handler'}], 'is_group': False, 'exceptions': []}]

What you think should happen instead?

Instead of running into the failure and being unable to complete the trigger task to queue up the asset-scheduled DAG run, the triggerer should successfully process the trigger's kwargs value, properly deserialize it, and complete the trigger function to trigger the DAG run for the asset-scheduled DAG.

All the details regarding the error and what should have been are mentioned in the previous section.

Operating System

MacOS

Deployment

Docker-Compose

Apache Airflow Provider(s)

apache-kafka, common-messaging

Versions of Apache Airflow Providers

apache-airflow-providers-apache-kafka==1.13.2
apache-airflow-providers-common-messaging==2.0.3

Official Helm Chart version

Not Applicable

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Custom Docker image file (Dockerfile) for Airflow image:

FROM apache/airflow:3.2.1-python3.14
USER root
# Build and install custom modules for Airflow
COPY --chown=airflow:root pyproject.toml .
COPY --chown=airflow:root requirements.txt .
COPY --chown=airflow:root scripts/airflow-init.sh .
RUN chown -R airflow:root logs
USER airflow
ARG AIRFLOW_HOME="/opt/airflow"
ARG VIRTUAL_ENV="/home/airflow/.local"
ARG PYTHONPATH="${VIRTUAL_ENV}/bin:${VIRTUAL_ENV}/lib/python3.14/site-packages"
ENV PYTHONPATH="${PYTHONPATH}:${AIRFLOW_HOME}/src:${AIRFLOW_HOME}/plugins"
# Install dependencies
RUN pip install -r requirements.txt

Docker Compose specifications only include the relevant services without the shared configurations:

x-airflow-common:
  # In order to add custom dependencies or upgrade provider distributions you can use your extended image.
  # Comment the image line, place your Dockerfile in the directory where you placed the docker-compose.yaml
  # and uncomment the "build" line below, Then run `docker-compose build` to build the images.
  # image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:latest}
  &airflow-common
  build: .
  env_file:
    - path: .env
      required: false
  environment: &airflow-common-env # Airflow configurations
    AIRFLOW_UID: 50000
    AIRFLOW_ENV: ${AIRFLOW_ENV:-prod}
    AIRFLOW__CORE__EXECUTOR: CeleryExecutor
    AIRFLOW__CORE__AUTH_MANAGER: airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager
    AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
    AIRFLOW__CORE__FERNET_KEY: ""
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: "true"
    AIRFLOW__CORE__LOAD_EXAMPLES: "false"
    AIRFLOW__CORE__EXECUTION_API_SERVER_URL: "http://airflow-apiserver:8080/execution/"
    AIRFLOW__CORE__LAZY_LOAD_PLUGINS: "false"
    AIRFLOW__CORE__LAZY_DISCOVER_PROVIDERS: "false"
    HOME_DIRECTORY: ${HOME_DIRECTORY:-/home/hayesadmin}
    AIRFLOW_USERNAME: ${AIRFLOW_USERNAME:-airflow}
    AIRFLOW_PASSWORD: ${AIRFLOW_PASSWORD:?error}
    _AIRFLOW_WWW_USER_USERNAME: ${_AIRFLOW_WWW_USER_USERNAME:-${AIRFLOW_USERNAME}}
    _AIRFLOW_WWW_USER_PASSWORD: ${_AIRFLOW_WWW_USER_PASSWORD:-${AIRFLOW_PASSWORD}}
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: ${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:-postgresql+psycopg2://${AIRFLOW_USERNAME}:${AIRFLOW_PASSWORD}@postgres/airflow}
    AIRFLOW__CELERY__RESULT_BACKEND: ${AIRFLOW__CELERY__RESULT_BACKEND:-db+postgresql://${AIRFLOW_USERNAME}:${AIRFLOW_PASSWORD}@postgres/airflow}
    # Airflow external database connection strings; Naming convention: AIRFLOW_CONN_{CONN_ID}
    # E.g. The following connection string can be accessed in the code using `conn_id="erp"`
    AIRFLOW_CONN_HWMS: ${AIRFLOW_CONN_HWMS:?error}
    AIRFLOW_CONN_CIWM: ${AIRFLOW_CONN_CIWM:?error}
    AIRFLOW_CONN_MATLIB: ${AIRFLOW_CONN_MATLIB:?error}
    AIRFLOW_CONN_ADMIN: ${AIRFLOW_CONN_ADMIN:?error}
    # yamllint disable rule:line-length
    # Use simple http server on scheduler for health checks
    # See https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/check-health.html#scheduler-health-check-server
    # yamllint enable rule:line-length
    AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: "true"
    # WARNING: Use _PIP_ADDITIONAL_REQUIREMENTS option ONLY for a quick checks
    # for other purpose (development, test and especially production usage) build/extend Airflow image.
    _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
    # The following line can be used to set a custom config file, stored in the local config folder
    AIRFLOW_CONFIG: /opt/airflow/config/airflow.cfg
    AIRFLOW_CONN_MY_KAFKA_CONFIG: kafka://kafka:9092?bootstrap.servers=kafka:9092&group.id=oms-kafka-group&auto.offset.reset=earliest
  volumes:
    - ${AIRFLOW_PROJ_DIR:-.}/src:/opt/airflow/src
    - ${HOME_DIRECTORY:-.}/oms-airflow-logs:/opt/airflow/logs
    - ${AIRFLOW_PROJ_DIR:-.}/config:/opt/airflow/config
    - ${AIRFLOW_PROJ_DIR:-.}/plugins:/opt/airflow/plugins
  user: "${AIRFLOW_UID:-50000}:0"
  depends_on: &airflow-common-depends-on
    redis:
      condition: service_healthy
    postgres:
      condition: service_healthy
    kafka:
      condition: service_healthy
    connect:
      condition: service_healthy

services:
  postgres:
    image: postgres:17.5
    command: -c 'max_connections=500'
    environment:
      POSTGRES_USER: ${AIRFLOW_USERNAME}
      POSTGRES_PASSWORD: ${AIRFLOW_PASSWORD:?error}
      POSTGRES_DB: airflow
    volumes:
      - postgres-db-volume:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 10s
      retries: 5
      start_period: 5s
    ports:
      - 6543:5432
    restart: unless-stopped

  redis:
    image: eqalpha/keydb
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "keydb-cli", "ping"]
      interval: 10s
      timeout: 30s
      retries: 50
      start_period: 10s
    restart: unless-stopped

  kafka:
    image: quay.io/debezium/kafka:3.4
    ports:
      - "9092:9093"
    volumes:
      - ~/debezium-kafka:/kafka/data
    environment:
      KAFKA_LOG_DIRS: /kafka/data
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://0.0.0.0:9093,CONTROLLER://kafka:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
    user: root
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list || exit 1",
        ]
      interval: 5s
      timeout: 10s
      retries: 6
      start_period: 10s
    restart: unless-stopped

  connect:
    image: quay.io/debezium/connect:3.4
    ports:
      - "8083:8083"
    volumes:
      - ~/debezium-kafka:/kafka/data
    environment:
      CONFIG_STORAGE_TOPIC: "debezium_connect_configs"
      OFFSET_STORAGE_TOPIC: "debezium_connect_offsets"
      STATUS_STORAGE_TOPIC: "debezium_connect_statuses"
      BOOTSTRAP_SERVERS: kafka:9092
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8083/"]
      interval: 10s
      timeout: 10s
      retries: 6
      start_period: 10s
    restart: unless-stopped
    depends_on:
      kafka:
        condition: service_healthy

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "9000:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: debezium-kafka
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: debezium-connect
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://connect:8083
      DYNAMIC_CONFIG_ENABLED: true
    restart: unless-stopped
    depends_on:
      kafka:
        condition: service_healthy
      connect-init:
        condition: service_completed_successfully

  airflow-apiserver:
    <<: *airflow-common
    command: api-server
    ports:
      - "8080:8080"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8080/api/v2/version"]
      interval: 10s
      timeout: 10s
      retries: 10
      start_period: 10s
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler
    ports:
      - "8974:8974"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8974/health"]
      interval: 10s
      timeout: 10s
      retries: 10
      start_period: 10s
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-dag-processor:
    <<: *airflow-common
    command: dag-processor
    healthcheck:
      test:
        [
          "CMD-SHELL",
          'airflow jobs check --job-type DagProcessorJob --hostname "$${HOSTNAME}"',
        ]
      interval: 10s
      timeout: 10s
      retries: 10
      start_period: 10s
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-worker:
    <<: *airflow-common
    command: celery worker
    healthcheck:
      # yamllint disable rule:line-length
      test:
        [
          "CMD-SHELL",
          'celery --app airflow.providers.celery.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}" || celery --app airflow.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"',
        ]
      interval: 10s
      timeout: 10s
      retries: 10
      start_period: 10s
    environment:
      <<: *airflow-common-env
      # Required to handle warm shutdown of the celery workers properly
      # See https://airflow.apache.org/docs/docker-stack/entrypoint.html#signal-propagation
      DUMB_INIT_SETSID: "0"
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-apiserver:
        condition: service_healthy
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: triggerer
    healthcheck:
      test:
        [
          "CMD-SHELL",
          'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"',
        ]
      interval: 10s
      timeout: 10s
      retries: 10
      start_period: 10s
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    command: airflow-init.sh
    environment:
      <<: *airflow-common-env
      _AIRFLOW_DB_MIGRATE: "true"
      _AIRFLOW_WWW_USER_CREATE: "true"
      _PIP_ADDITIONAL_REQUIREMENTS: ""
    user: "${AIRFLOW_UID:-50000}:0"
    depends_on:
      - postgres
      - redis

  # You can enable flower by adding "--profile flower" option e.g. docker-compose --profile flower up
  # or by explicitly targeted on the command line e.g. docker-compose up flower.
  # See: https://docs.docker.com/compose/profiles/
  flower:
    <<: *airflow-common
    command: celery flower
    ports:
      - "5555:5555"
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:5555/"]
      interval: 10s
      timeout: 10s
      retries: 5
      start_period: 10s
    restart: unless-stopped
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

Note that some of the env values are set as required, so if you are trying to replicate the same exact Airflow Docker environment using the above two Docker files (Docker image and Docker Compose), you can take a look at which environment variables for Docker Compose are set as required and add those with random values to your local .env file in the same directory as these Docker files.

Docker commands used:

1. docker compose down --remove-orphans
2. docker compose build
3. docker compose up -d

In order to successfully run this customer Airflow Docker image, you will need the airflow-init.sh in the scripts folder in the folder where the Docker image and Docker Compose files are in:

if [[ -z "${AIRFLOW_UID}" ]]; then
	echo
	echo -e "\033[1;33mWARNING!!!: AIRFLOW_UID not set!\e[0m"
	echo "If you are on Linux, you SHOULD follow the instructions below to set "
	echo "AIRFLOW_UID environment variable, otherwise files will be owned by root."
	echo "For other operating systems you can get rid of the warning with manually created .env file:"
	echo "    See: https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#setting-the-right-airflow-user"
	echo
	export AIRFLOW_UID=$(id -u)
fi
one_meg=1048576
mem_available=$(($(getconf _PHYS_PAGES) * $(getconf PAGE_SIZE) / one_meg))
cpus_available=$(grep -cE 'cpu[0-9]+' /proc/stat)
disk_available=$(df / | tail -1 | awk '{print $4}')
warning_resources="false"
if (( mem_available < 4000 )) ; then
	echo
	echo -e "\033[1;33mWARNING!!!: Not enough memory available for Docker.\e[0m"
	echo "At least 4GB of memory required. You have $$(numfmt --to iec $$((mem_available * one_meg)))"
	echo
	warning_resources="true"
fi
if (( cpus_available < 2 )); then
	echo
	echo -e "\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\e[0m"
	echo "At least 2 CPUs recommended. You have $${cpus_available}"
	echo
	warning_resources="true"
fi
if (( disk_available < one_meg * 10 )); then
	echo
	echo -e "\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\e[0m"
	echo "At least 10 GBs recommended. You have $$(numfmt --to iec $$((disk_available * 1024 )))"
	echo
	warning_resources="true"
fi
if [[ $${warning_resources} == "true" ]]; then
	echo
	echo -e "\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\e[0m"
	echo "Please follow the instructions to increase amount of resources available:"
	echo "   https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#before-you-begin"
	echo
fi
echo
echo "Creating missing opt dirs if missing:"
echo
mkdir -v -p /opt/airflow/{logs,src,plugins,config}
echo
echo "Airflow version:"
/entrypoint airflow version
echo
echo "Files in shared volumes:"
echo
ls -la /opt/airflow/{logs,src,plugins,config}
echo
echo "Running airflow config list to create default config file if missing."
echo
/entrypoint airflow config list >/dev/null
echo
echo "Files in shared volumes:"
echo
ls -la /opt/airflow/{logs,src,plugins,config}

And finally, the custom Airflow config file should be under the config/airflow.cfg. The default generated config should be fine (using the airflow config default command).

Anything else?

Happens every time.
Reserializing does not help, as it will constantly output the erroneously serialized kwargs value for the trigger into the metadata database for Airflow.
Always end up with the same apply_function() got an unexpected keyword argument 'Encoding.VAR'.

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

[boring-cyborg]

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.

[shaealh]

Taking a look

[shaealh]

Opened this PR