[Console] Report Flink and Spark job lineage to Gravitino via OpenLineage

pom.xml

@@ -125,6 +125,7 @@
         <commons-lang3.version>3.12.0</commons-lang3.version>
         <enumeratum.version>1.6.1</enumeratum.version>
         <assertj.version>3.23.1</assertj.version>
+        <openlineage.version>1.29.0</openlineage.version>
 
         <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
         <maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
@@ -142,7 +143,7 @@
         <build-helper-maven-plugin.version>3.3.0</build-helper-maven-plugin.version>
         <flatten-maven-plugin.version>1.6.0</flatten-maven-plugin.version>
         <streampark.shaded.package>org.apache.streampark.shaded</streampark.shaded.package>
-        <httpclient5.version>5.1</httpclient5.version>
+        <httpclient5.version>5.4.2</httpclient5.version>
         <lombok.version>1.18.24</lombok.version>
         <jupiter.version>5.9.1</jupiter.version>
         <mockito.version>3.4.6</mockito.version>
@@ -342,6 +343,12 @@
                 <version>${commons-collections4.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.openlineage</groupId>
+                <artifactId>openlineage-java</artifactId>
+                <version>${openlineage.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>org.apache.httpcomponents.client5</groupId>
                 <artifactId>httpclient5</artifactId>
@@ -354,6 +361,21 @@
                 <version>${httpclient5.version}</version>
             </dependency>
 
+            <!-- httpclient5 5.4.2 requires httpcore5 5.4.x; without pinning this explicitly, an
+                 unrelated older transitive declaration wins dependency mediation and httpclient5
+                 fails at runtime with NoSuchMethodError against the mismatched httpcore5. -->
+            <dependency>
+                <groupId>org.apache.httpcomponents.core5</groupId>
+                <artifactId>httpcore5</artifactId>
+                <version>5.4.3</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.httpcomponents.core5</groupId>
+                <artifactId>httpcore5-h2</artifactId>
+                <version>5.4.3</version>
+            </dependency>
+
             <dependency>
                 <groupId>org.apache.hadoop</groupId>
                 <artifactId>hadoop-client-api</artifactId>

streampark-console/streampark-console-service/pom.xml

@@ -295,6 +295,12 @@
             <version>${commons-email.version}</version>
         </dependency>
 
+        <!-- lineage: emits OpenLineage RunEvents to Gravitino's POST /api/lineage -->
+        <dependency>
+            <groupId>io.openlineage</groupId>
+            <artifactId>openlineage-java</artifactId>
+        </dependency>
+
         <!-- jwt -->
         <dependency>
             <groupId>com.auth0</groupId>

streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql

@@ -328,6 +328,10 @@ insert into `t_setting` values (12, 'docker.register.user', null, 'Docker Regist
 insert into `t_setting` values (13, 'docker.register.password', null, 'Docker Register Password', 'Docker container service authentication password', 1);
 insert into `t_setting` values (14, 'docker.register.namespace', null, 'Docker namespace', 'Namespace for docker image used in docker building env and target image register', 1);
 insert into `t_setting` values (15, 'ingress.mode.default', null, 'Ingress domain address', 'Automatically generate an nginx-based ingress by passing in a domain name', 1);
+insert into `t_setting` values (16, 'lineage.gravitino.address', null, 'Gravitino Address', 'Base URL of the Gravitino server lineage events are reported to, e.g. http://host:8090', 1);
+insert into `t_setting` values (17, 'lineage.gravitino.token', null, 'Gravitino Auth Token', 'Bearer token forwarded to Gravitino, required once Gravitino oauth authentication is enabled', 1);
+insert into `t_setting` values (18, 'lineage.gravitino.namespace', null, 'Gravitino Lineage Namespace', 'OpenLineage job/dataset namespace StreamPark reports under', 1);
+insert into `t_setting` values (19, 'lineage.flink.native.listener.enable', 'true', 'Enable Flink Native OpenLineage Listener', 'Whether to also inject the official openlineage-flink job-status-changed-listener config; only takes effect once Gravitino Address is set', 2);
 -- ----------------------------
 -- Records of t_user
 -- ----------------------------

streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql

@@ -277,6 +277,10 @@ insert into "public"."t_setting" values (12, 'docker.register.user', null, 'Dock
 insert into "public"."t_setting" values (13, 'docker.register.password', null, 'Docker Register Password', 'Docker container service authentication password', 1);
 insert into "public"."t_setting" values (14, 'docker.register.namespace', null, 'Docker namespace', 'Namespace for docker image used in docker building env and target image register', 1);
 insert into "public"."t_setting" values (15, 'ingress.mode.default', null, 'Ingress domain address', 'Automatically generate an nginx-based ingress by passing in a domain name', 1);
+insert into "public"."t_setting" values (16, 'lineage.gravitino.address', null, 'Gravitino Address', 'Base URL of the Gravitino server lineage events are reported to, e.g. http://host:8090', 1);
+insert into "public"."t_setting" values (17, 'lineage.gravitino.token', null, 'Gravitino Auth Token', 'Bearer token forwarded to Gravitino, required once Gravitino oauth authentication is enabled', 1);
+insert into "public"."t_setting" values (18, 'lineage.gravitino.namespace', null, 'Gravitino Lineage Namespace', 'OpenLineage job/dataset namespace StreamPark reports under', 1);
+insert into "public"."t_setting" values (19, 'lineage.flink.native.listener.enable', 'true', 'Enable Flink Native OpenLineage Listener', 'Whether to also inject the official openlineage-flink job-status-changed-listener config; only takes effect once Gravitino Address is set', 2);
 
 -- ----------------------------
 -- Records of t_user

streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql

@@ -99,6 +99,7 @@ create table `t_flink_app` (
   `default_mode_ingress` text collate utf8mb4_general_ci,
   `tags` varchar(500) default null,
   `hadoop_user` varchar(64) collate utf8mb4_general_ci default null,
+  `lineage_enable` tinyint default 0,
   primary key (`id`) using btree,
   key `inx_job_type` (`job_type`) using btree,
   key `inx_track` (`tracking`) using btree,
@@ -616,6 +617,7 @@ create table `t_spark_app` (
   `k8s_executor_pod_template` text collate utf8mb4_general_ci,
   `k8s_hadoop_integration` tinyint default 0,
   `hadoop_user` varchar(64) collate utf8mb4_general_ci default null,
+  `lineage_enable` tinyint default 0,
   `restart_size` int default null,
   `restart_count` int default null,
   `state` int default null,

streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql

@@ -252,7 +252,8 @@ create table "public"."t_flink_app" (
   "ingress_template" text collate "pg_catalog"."default",
   "default_mode_ingress" text collate "pg_catalog"."default",
   "tags" varchar(500) collate "pg_catalog"."default",
-  "hadoop_user" varchar(63) collate "pg_catalog"."default"
+  "hadoop_user" varchar(63) collate "pg_catalog"."default",
+  "lineage_enable" boolean default false
 )
 ;
 alter table "public"."t_flink_app" add constraint "t_flink_app_pkey" primary key ("id");

streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/3.0.0.sql

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use streampark;
+
+set names utf8mb4;
+set foreign_key_checks = 0;
+
+-- ----------------------------
+-- lineage: t_setting
+-- ----------------------------
+insert into `t_setting` values (16, 'lineage.gravitino.address', null, 'Gravitino Address', 'Base URL of the Gravitino server lineage events are reported to, e.g. http://host:8090', 1);
+insert into `t_setting` values (17, 'lineage.gravitino.token', null, 'Gravitino Auth Token', 'Bearer token forwarded to Gravitino, required once Gravitino oauth authentication is enabled', 1);
+insert into `t_setting` values (18, 'lineage.gravitino.namespace', null, 'Gravitino Lineage Namespace', 'OpenLineage job/dataset namespace StreamPark reports under', 1);
+insert into `t_setting` values (19, 'lineage.flink.native.listener.enable', 'true', 'Enable Flink Native OpenLineage Listener', 'Whether to also inject the official openlineage-flink job-status-changed-listener config; only takes effect once Gravitino Address is set', 2);
+
+-- ----------------------------
+-- lineage: t_flink_app / t_spark_app
+-- ----------------------------
+alter table `t_flink_app`
+add column `lineage_enable` tinyint default 0;
+
+alter table `t_spark_app`
+add column `lineage_enable` tinyint default 0;

streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/3.0.0.sql

@@ -0,0 +1,35 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements.  See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License.  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+-- ----------------------------
+-- lineage: t_setting
+-- ----------------------------
+insert into "public"."t_setting" values (16, 'lineage.gravitino.address', null, 'Gravitino Address', 'Base URL of the Gravitino server lineage events are reported to, e.g. http://host:8090', 1);
+insert into "public"."t_setting" values (17, 'lineage.gravitino.token', null, 'Gravitino Auth Token', 'Bearer token forwarded to Gravitino, required once Gravitino oauth authentication is enabled', 1);
+insert into "public"."t_setting" values (18, 'lineage.gravitino.namespace', null, 'Gravitino Lineage Namespace', 'OpenLineage job/dataset namespace StreamPark reports under', 1);
+insert into "public"."t_setting" values (19, 'lineage.flink.native.listener.enable', 'true', 'Enable Flink Native OpenLineage Listener', 'Whether to also inject the official openlineage-flink job-status-changed-listener config; only takes effect once Gravitino Address is set', 2);
+
+-- ----------------------------
+-- lineage: t_flink_app
+-- ----------------------------
+alter table "public"."t_flink_app"
+add column "lineage_enable" boolean default false;
+
+-- Note: t_spark_app has no PostgreSQL schema definition anywhere in this repository (pre-existing
+-- gap, confirmed absent from pgsql-schema.sql; not introduced by this change and not fixed here,
+-- out of scope for a lineage feature migration). The corresponding "lineage_enable" column for
+-- t_spark_app is added only in upgrade/mysql/3.0.0.sql; add it here once that gap is fixed.

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/LineageConfig.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.bean;
+
+import org.apache.commons.lang3.StringUtils;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Gravitino lineage configuration. {@link #enabled()} is the single gate every lineage call site
+ * checks before doing any work: it is only true when an operator has actually filled in a
+ * Gravitino address, so a fresh install with these settings left blank injects nothing and emits
+ * nothing.
+ */
+@Getter
+@Setter
+public class LineageConfig {
+
+    /**
+     * Namespace used when an operator has not set one. It ends up inside emitted event payloads and
+     * inside injected job configuration, so both paths must agree on it — hence one definition.
+     */
+    public static final String DEFAULT_NAMESPACE = "streampark";
+
+    /**
+     * Gravitino's lineage-ingest path, appended to {@link #gravitinoAddress}. Written into the
+     * Flink and Spark listener configuration as well as used by the Console's own emitter, and all
+     * three must address the same endpoint — hence one definition.
+     */
+    public static final String LINEAGE_ENDPOINT_PATH = "/api/lineage";
+
+    /** Gravitino base URL, e.g. {@code http://192.168.10.132:8090}. */
+    private String gravitinoAddress;
+
+    /** Bearer token forwarded as-is to Gravitino's {@code /api/lineage}; required once oauth is enabled there. */
+    private String gravitinoToken;
+
+    /** OpenLineage job/dataset namespace StreamPark reports under. */
+    private String gravitinoNamespace;
+
+    /** Whether to also inject the official {@code openlineage-flink} job-status-changed-listener config. */
+    private boolean flinkNativeListenerEnable;
+
+    public boolean enabled() {
+        return StringUtils.isNotBlank(gravitinoAddress);
+    }
+
+    /** The configured namespace, or {@link #DEFAULT_NAMESPACE} when none was set. */
+    public String namespaceOrDefault() {
+        return StringUtils.defaultIfBlank(gravitinoNamespace, DEFAULT_NAMESPACE);
+    }
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkApplication.java

@@ -140,6 +140,9 @@ public class FlinkApplication extends BaseEntity implements ApplicationEntitySup
      */
     private Boolean k8sHadoopIntegration;
 
+    /** Whether to report OpenLineage data lineage for this application to Gravitino. */
+    private Boolean lineageEnable;
+
     private Integer state;
     /**
      * task release status

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplication.java

@@ -147,6 +147,9 @@ public class SparkApplication extends BaseEntity implements ApplicationEntitySup
     @TableField("HADOOP_USER")
     private String hadoopUser;
 
+    /** Whether to report OpenLineage data lineage for this application to Gravitino. */
+    private Boolean lineageEnable;
+
     /** max restart retries after job failed */
     @TableField(updateStrategy = FieldStrategy.IGNORED)
     private Integer restartSize;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/request/flink/FlinkAppCreateRequest.java

@@ -125,4 +125,6 @@ public class FlinkAppCreateRequest implements Serializable {
     private Boolean k8sHadoopIntegration;
 
     private String serviceAccount;
+
+    private Boolean lineageEnable;
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/request/spark/SparkAppCreateRequest.java

@@ -86,6 +86,8 @@ public class SparkAppCreateRequest implements Serializable {
 
     private String hadoopUser;
 
+    private Boolean lineageEnable;
+
     private Integer restartSize;
 
     private Long alertId;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/response/flink/FlinkAppResponse.java

@@ -152,6 +152,8 @@ public class FlinkAppResponse implements Serializable {
 
     private Boolean k8sHadoopIntegration;
 
+    private Boolean lineageEnable;
+
     private JobsOverview.Task overview;
 
     private String teamResource;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/response/spark/SparkAppResponse.java

@@ -91,6 +91,8 @@ public class SparkAppResponse implements Serializable {
 
     private String hadoopUser;
 
+    private Boolean lineageEnable;
+
     private Integer restartSize;
 
     private Integer restartCount;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/GravitinoLineageService.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.service;
+
+import org.apache.streampark.console.core.entity.FlinkApplication;
+import org.apache.streampark.flink.core.lineage.LineagePipeline;
+
+import java.util.List;
+
+/**
+ * Reports Flink job table-level lineage to Gravitino's {@code POST /api/lineage} as OpenLineage
+ * {@code RunEvent}s.
+ *
+ * <p>Every method here is fail-open by contract: a disabled switch, an unconfigured Gravitino
+ * address, or any failure while talking to Gravitino is logged and swallowed, never thrown. This
+ * runs on the job submission and state-watching paths, where a lineage gap must never affect the
+ * job itself.
+ *
+ * <p>{@link #trackAndEmitStart} and {@link #emitTerminal} are a pair: a successful start call
+ * remembers the run in memory so the later terminal call (driven by {@code FlinkAppHttpWatcher}'s
+ * state polling) knows what to close out, without needing the caller to thread pipeline data
+ * through the whole state-watching path. This tracking is in-memory only — it does not survive a
+ * Console restart, so a run whose job finishes while Console is down never gets its COMPLETE/FAIL
+ * event. That is a deliberate, bounded scope decision (see the implementation), not an oversight.
+ */
+public interface GravitinoLineageService {
+
+    /**
+     * Called once, right after a Flink SQL job's submission succeeds. No-ops if {@code pipelines}
+     * is empty (extraction found nothing, or lineage is disabled for this application/globally).
+     *
+     * @param application the just-started application (its id keys the in-memory pending-run
+     *     tracking consumed by {@link #emitTerminal})
+     * @param flinkJobIdHex the Flink JobID this run was submitted with
+     * @param pipelines the pipelines resolved from the job's SQL; safe to pass an empty list
+     */
+    void trackAndEmitStart(FlinkApplication application, String flinkJobIdHex, List<LineagePipeline> pipelines);
+
+    /**
+     * Called when {@code FlinkAppHttpWatcher} observes an application transition into a terminal
+     * state. No-ops if no pending run is tracked for {@code appId} (lineage was never started for
+     * this run, or it was already closed out).
+     *
+     * @param appId the application id
+     * @param success {@code true} to emit COMPLETE, {@code false} to emit FAIL
+     */
+    void emitTerminal(Long appId, boolean success);
+}