如何在Airflow的on_failure_callback中捕获Java应用的异常？

方案1：自定义Operator捕获Pod日志并解析异常 #

KubernetesPodOperator抛出的AirflowException仅为容器退出的包装，实际Java异常存在Pod日志中。通过继承Operator并重写execute方法，拉取Pod日志解析出真实异常后重新抛出，让回调能获取到详情。

代码实现 #

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
from airflow.exceptions import AirflowException
import re

class CustomBeamKubernetesPodOperator(KubernetesPodOperator):
    def execute(self, context):
        try:
            # 执行父类核心逻辑
            super().execute(context)
        except AirflowException as e:
            # 获取Pod的名称与命名空间
            pod_name = self.pod_name or self.generate_pod_name(context)
            namespace = self.namespace or KubernetesHook.get_default_namespace()
            
            # 拉取Pod完整运行日志
            k8s_hook = KubernetesHook(conn_id=self.kubernetes_conn_id)
            pod_logs = k8s_hook.get_pod_logs(pod_name=pod_name, namespace=namespace)
            
            # 匹配并提取Java异常栈信息（可根据实际日志格式调整规则）
            error_lines = []
            in_exception_block = False
            for line in pod_logs.split('\n'):
                if 'Exception:' in line or 'Error:' in line:
                    in_exception_block = True
                    error_lines.append(line)
                elif in_exception_block and line.startswith('\tat'):
                    error_lines.append(line)
                elif in_exception_block and not line.startswith('\t'):
                    break
            
            if error_lines:
                actual_error = '\n'.join(error_lines)
                # 包装包含真实异常的新异常抛出
                raise AirflowException(f"任务失败，Java运行时异常:\n{actual_error}") from e
            # 未解析到异常则抛出原错误
            raise e

回调中获取异常 #

def task_failure(context):
    exception = context.get('exception')
    if exception:
        # 此处可直接获取包含Java实际异常的详情
        print(f"失败详情: {str(exception)}")
        # 可在此扩展告警推送、错误日志存储等逻辑

方案2：通过Dataflow API查询作业错误详情 #

Beam Dataflow作业会在对应云平台（以GCP为例）留存完整作业状态，可通过云API直接拉取作业的失败原因。

实现步骤 #

1. 在Java应用启动Dataflow作业时，主动将作业ID输出到日志（例如System.out.println("Dataflow Job ID: " + job.getId());）。
2. 在回调中提取日志中的作业ID，调用Dataflow API获取错误详情：

from google.cloud import dataflow_v1beta3
import re

def task_failure(context):
    # 从XCom获取自定义Operator存储的Pod日志（需在Operator中添加xcom_push逻辑）
    pod_logs = context['ti'].xcom_pull(task_ids=context['ti'].task_id, key='beam_pod_logs')
    if not pod_logs:
        return
    
    # 提取Dataflow作业ID
    job_id_match = re.search(r'Dataflow Job ID: (\w+)', pod_logs)
    if not job_id_match:
        return
    
    job_id = job_id_match.group(1)
    # 初始化Dataflow客户端
    client = dataflow_v1beta3.JobsV1Beta3Client()
    # 替换为作业实际部署的区域
    job = client.get_job(job_id=job_id, location='us-central1')
    
    if job.current_state == dataflow_v1beta3.JobState.JOB_STATE_FAILED:
        error_detail = job.state_message
        print(f"Dataflow作业失败原因: {error_detail}")

方案3：标准化Java日志输出 #

让Java应用将异常信息输出到标准错误流（stderr），并保持格式统一。KubernetesPodOperator默认会捕获容器的stderr/stdout，开启get_logs=True（默认开启）后，日志会被Airflow记录，可直接在自定义Operator中提取。

注意事项 #

• 确保Airflow使用的服务账号有权限访问Kubernetes Pod日志（Composer默认服务账号已具备该权限，自定义账号需配置对应RBAC规则）。
• Dataflow API方案需确保Airflow环境具备调用云API的权限（例如GCP环境中需赋予Composer服务账号roles/dataflow.viewer权限）。

内容的提问来源于stack exchange，提问作者Maria Dorohin