Flink JobStatusChangedListener把作业状态变化 + Lineage 发到 DataHub / OpenLineage

1、它是什么？何时触发？

• 你实现的是：JobStatusChangedListener

• 触发时机：应用（作业）状态每次变化都会触发

  • 比如 CREATED / RUNNING / FAILING / FAILED / CANCELLING / CANCELED / FINISHED 等（以 Flink 的 job status 为准）

• 血缘信息（lineage）在哪里？

  • JobCreatedEvent中包含 source/sink lineage 信息
  • 你在onEvent(JobStatusChangedEvent event)里收到事件，遇到JobCreatedEvent时可以解析并上报

2、插件怎么做：3 步走（和 FailureEnricher 很像）

要实现一个自定义 JobStatusChangedListener 插件，你需要：

1. 实现JobStatusChangedListener
2. 实现JobStatusChangedListenerFactory
3. 使用 Java SPI 注册工厂类：
   创建文件
   META-INF/services/org.apache.flink.core.execution.JobStatusChangedListenerFactory
   内容是你的 Factory 全类名

打成 jar 后放入 Flink 的 plugins 目录（独立 classloader 隔离）：

$FLINK_HOME/plugins/ job-status-changed-listener/ flink-lineage-listener.jar

3、最小示例代码（可直接改成 DataHub/OpenLineage 上报）

3.1 Factory 示例

packagecom.yourcompany.flink.lineage;publicclassMyJobStatusChangedListenerFactoryimplementsJobStatusChangedListenerFactory{@OverridepublicJobStatusChangedListenercreateListener(Contextcontext){returnnewMyJobStatusChangedListener(context);}}

3.2 Listener 示例（接收事件并处理 JobCreatedEvent）

packagecom.yourcompany.flink.lineage;publicclassMyJobStatusChangedListenerimplementsJobStatusChangedListener{privatefinalJobStatusChangedListenerFactory.Contextcontext;publicMyJobStatusChangedListener(JobStatusChangedListenerFactory.Contextcontext){this.context=context;}@OverridepublicvoidonEvent(JobStatusChangedEventevent){// 1) 所有状态变化都会到这里// 2) lineage 在 JobCreatedEvent 里if(eventinstanceofJobCreatedEvent){JobCreatedEventcreated=(JobCreatedEvent)event;// 伪代码：从 created 中提取 source/sink lineage// LineageInfo lineage = created.getLineageInfo();// sendToDatahubOrOpenLineage(lineage);}else{// 你也可以上报运行/失败/取消等状态，用于血缘系统里的运行实例关联// sendJobStatus(event.getNewStatus(), event.getJobId(), ...)}}}

3.3 SPI 文件内容

文件路径：

META-INF/services/org.apache.flink.core.execution.JobStatusChangedListenerFactory

文件内容（一个工厂类名一行）：

com.yourcompany.flink.lineage.MyJobStatusChangedListenerFactory

4、配置：不配就不启动（关键）

Flink 启动时加载 JobStatusChangedListener 插件，但是否启用由配置决定：

execution.job-status-changed-listeners = com.yourcompany.flink.lineage.MyJobStatusChangedListenerFactory

注意：

• 这里配置的是Factory 的类名（你给的例子也是 factory）
• 如果该配置为空：不会启动任何 listener

5）实践建议：做成“血缘上报”生产可用版本

• 只在 JobCreatedEvent 做血缘解析：避免每次状态变化都做重逻辑
• 异步上报 + 超时兜底：上报系统不可用时别拖垮 JM 线程
• 幂等/去重：同一个 jobId + runId（或提交时间）做幂等 key，避免重复写血缘
• 关联运行实例：把 RUNNING/FAILED/FINISHED 状态也上报，用于 lineage 系统展示“本次运行”
• 多环境标识：在上报里附加 clusterId、namespace、env、tenant 等标签，方便血缘平台分组