Airflow PubSubHook acknowledge方法metadata参数异常求助

问题概述 #

在Composer 1.19.12（Airflow 2.3.3）环境下，使用PubSubPull Operator的回调方法手动确认PubSub消息时，遇到与metadata参数相关的错误：

• 传入metadata=[]：报错TypeError: __call__() got an unexpected keyword argument 'metadata'
• 传入metadata=None：报错TypeError: 'NoneType' object is not iterable
• 省略metadata参数：仍出现上述第一个错误

解决方案 #

关键修复点 #

1. 移除手动传递的metadata参数：当前版本的Airflow PubSubHook与底层Google PubSub客户端存在参数兼容问题，无需手动传入metadata，由Hook自动处理身份验证相关的元数据。
2. 实例化Retry对象：不要直接传入Retry类，需创建具体的Retry实例（如设置超时时间）。
3. 复用PubSubHook实例：避免在回调循环中重复创建Hook实例，提升性能并减少潜在问题。

修改后的完整代码 #

from __future__ import annotations
import os
from datetime import datetime
import base64
import airflow
from airflow import DAG
import json
from airflow.providers.google.cloud.operators.pubsub import (
    PubSubCreateSubscriptionOperator,
    PubSubPullOperator,
)
from airflow.providers.google.cloud.sensors.pubsub import PubSubPullSensor
from airflow.providers.google.cloud.hooks.pubsub import PubSubHook, Retry
from airflow.utils.trigger_rule import TriggerRule

ENV_ID = "Dev"
PROJECT_ID = "abcdef"
DAG_ID = "DataPullDag_1"
TOPIC_ID = "alert_topic_jp"
SNOW_SUBSCRIPTION = "alert_subscription_jp"

# 提前创建PubSubHook实例，避免重复初始化
pubsub_hook = PubSubHook()
# 实例化Retry对象，设置超时时间
retry_config = Retry(deadline=10)

def print_ack_messages(pulled_messages, context):
    for idx, m in enumerate(pulled_messages):
        data = m.message.data.decode('utf-8')
        print(f'################----------{data}')     
        data_json_dict = json.loads(data)     
        print(
            f"AckID: {m.ack_id}, incident_id: {data_json_dict['incident']['incident_id']} "
            f"scoping_project_id: {data_json_dict['incident']['scoping_project_id']} "
            f"resource_name: {data_json_dict['incident']['resource_name']} "
            f"summary: {data_json_dict['incident']['summary']}"
        )
        # 确认消息
        ack_id_list = [m.ack_id]
        print(type(ack_id_list))
        if idx == 0:
            # 移除metadata参数，使用预创建的Hook和Retry实例
            pubsub_hook.acknowledge(
                subscription=SNOW_SUBSCRIPTION,
                project_id=PROJECT_ID,
                ack_ids=ack_id_list,
                retry=retry_config,
                timeout=10
            )
            print(f"Successfully acknowledged incident_id: {data_json_dict['incident']['incident_id']}")


with DAG(
    DAG_ID,
    schedule_interval='@once',
    start_date=airflow.utils.dates.days_ago(0),
    catchup=False,
) as dag:
    subscribe_task = PubSubCreateSubscriptionOperator(
        task_id="subscribe_task",
        project_id=PROJECT_ID,
        topic=TOPIC_ID,
        subscription=SNOW_SUBSCRIPTION
    )
    
    subscription = subscribe_task.output

    pull_messages_operator = PubSubPullOperator(
        task_id="pull_messages_operator",
        ack_messages=False,
        project_id=PROJECT_ID,       
        messages_callback=print_ack_messages,
        subscription=subscription,
        max_messages=50,
    )

    subscribe_task >> pull_messages_operator

错误原因分析 #

1. 参数兼容问题：当前版本的Airflow PubSubHook在调用底层Google PubSub客户端的acknowledge方法时，错误地传递了metadata参数，但该客户端方法并不接受此参数，导致第一个错误。
2. None类型迭代错误：当传入metadata=None时，底层客户端尝试将其转换为元组进行拼接，但None无法被迭代，触发第二个错误。
3. Retry类未实例化：直接传入Retry类而非实例，会导致后续调用时的参数类型错误（虽未在当前报错中体现，但属于潜在问题）。

内容的提问来源于stack exchange，提问作者Despicable me