rahul-madaan opened a new pull request, #39428: URL: https://github.com/apache/beam/pull/39428
Adds `sdks/java/extensions/openlineage`, a new extension that emits [OpenLineage](https://openlineage.io) events describing the datasets a Beam pipeline reads and writes, so OpenLineage consumers (Marquez, Atlan, and others) can build lineage graphs for Beam jobs on any runner. addresses #39427 ### Design The extension deliberately mirrors the architecture of the official OpenLineage Spark and Flink integrations so it is familiar to both communities: - **Configuration** extends `io.openlineage.client.OpenLineageConfig` (like `SparkOpenLineageConfig`/`FlinkOpenLineageConfig`) and resolves with the Spark integration's precedence: pipeline options > `openlineage.yml` > `OPENLINEAGE__` env vars. Transport, disabled facets, dataset resolvers and circuit breaker all reuse the standard openlineage-java config classes; no secrets flow through pipeline options. - **`OpenLineageRunner`** wraps any runner (`--runner=OpenLineageRunner --openLineageDelegateRunner=FlinkRunner`): extracts datasets from the pipeline graph via classpath-guarded per-IO visitors (PubsubIO typed, IcebergIO reflective since its getters are package-private; extensible through the public `LineageProvider` interface), mints a UUIDv7 run id at submission and propagates it via serialized options, emits START and terminal COMPLETE/ABORT/FAIL (with an `errorMessage` facet on failure), and supports a parent run facet for orchestrator-launched pipelines. - **`OpenLineageJobTracker`** emits periodic RUNNING events on the Flink integration's `trackingIntervalInSeconds` cadence (default 60s) and sweeps lineage reported through the Beam `Lineage` metrics API at each tick and at terminal state. - **`OpenLineageLineage`** implements the `LineageBase` plugin (#36790, `--lineageType`) to capture lineage live inside worker JVMs for long-running streaming jobs — where runner metrics are not readable mid-run — teeing every FQN back into the metrics store so existing metrics-based consumers keep working unchanged. - **Dataset naming strictly follows the [OpenLineage naming conventions](https://openlineage.io/docs/spec/naming/)**. Iceberg datasets use the physical table location with the catalog table identity attached as a TABLE symlink, matching the Spark integration's `IcebergHandler`. Systems without a spec entry pass through with a conservative generic mapping. Emission can never fail a pipeline: every hook swallows exceptions and emission runs inside the configured OpenLineage circuit breaker. ### Testing - 26 unit tests: FQN-to-naming-convention conformance for every spec-listed system, graph extraction against real PubsubIO and real IcebergIO (Hadoop catalog with actual `loadTable()` location resolution, plus graceful fallback when a REST catalog is unreachable at submit time), end-to-end runner and plugin pipelines on the direct runner asserting the full emitted event sequence via a file transport, parent-facet and disabled-flag behavior, and config precedence. - Validated end-to-end against a live OpenLineage consumer: START/RUNNING/COMPLETE events from this module were ingested and produced the expected lineage assets. - `:sdks:java:extensions:openlineage:build` (checkstyle, spotless, SpotBugs, javadoc, tests) and `rat` pass. New dependency: `io.openlineage:openlineage-java` (Apache-2.0), scoped to this module only. GCP and Iceberg IO modules are compile-only; their visitors activate at runtime only when the corresponding IO is already on the user's classpath. ------------------------ Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [x] Mention the appropriate issue in your description (addresses #39427). - [x] Update `CHANGES.md` with noteworthy changes. - [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
