Add OpenLineage extension for Java SDK

CHANGES.md

@@ -69,6 +69,7 @@
 ## New Features / Improvements
 
 * (Python) Removed the `envoy-data-plane` (and transitive `betterproto`) dependency; `EnvoyRateLimiter` now uses a small vendored protobuf definition instead, resolving dependency conflicts for downstream projects ([#37854](https://github.com/apache/beam/issues/37854)).
+* Added an OpenLineage extension (`sdks/java/extensions/openlineage`) that emits OpenLineage events describing the datasets a pipeline reads and writes, via a wrapper runner (`--runner=OpenLineageRunner`) and a pluggable lineage backend (`--lineageType`) (Java) ([#39427](https://github.com/apache/beam/issues/39427)).
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Breaking Changes

build.gradle.kts

@@ -299,6 +299,7 @@ tasks.register("javaPreCommit") {
   dependsOn(":sdks:java:extensions:join-library:build")
   dependsOn(":sdks:java:extensions:kryo:build")
   dependsOn(":sdks:java:extensions:ml:build")
+  dependsOn(":sdks:java:extensions:openlineage:build")
   dependsOn(":sdks:java:extensions:protobuf:build")
   dependsOn(":sdks:java:extensions:python:build")
   dependsOn(":sdks:java:extensions:sbe:build")

sdks/java/extensions/openlineage/build.gradle

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+  automaticModuleName: 'org.apache.beam.sdk.extensions.openlineage',
+  // matches sdks/java/io/iceberg, which this module inspects at test time
+  requireJavaVersion: JavaVersion.VERSION_11,
+)
+
+description = "Apache Beam :: SDKs :: Java :: Extensions :: OpenLineage"
+ext.summary = "Emit OpenLineage events for Beam pipelines."
+
+def openlineage_version = "1.51.0"
+
+dependencies {
+  implementation project(path: ":sdks:java:core", configuration: "shadow")
+  implementation "io.openlineage:openlineage-java:$openlineage_version"
+  implementation library.java.jackson_annotations
+  implementation library.java.jackson_core
+  implementation library.java.joda_time
+  implementation library.java.slf4j_api
+  implementation library.java.vendored_guava_32_1_2_jre
+  // IO modules are compile-only: transform-specific lineage visitors activate at runtime only
+  // when the corresponding IO is already on the user's classpath.
+  compileOnly project(path: ":sdks:java:io:google-cloud-platform")
+  compileOnly project(path: ":sdks:java:io:iceberg")
+  // keep in sync with iceberg_version in sdks/java/io/iceberg/build.gradle
+  compileOnly "org.apache.iceberg:iceberg-api:1.10.0"
+  testImplementation library.java.junit
+  testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+  testImplementation project(path: ":sdks:java:io:google-cloud-platform")
+  testImplementation project(path: ":sdks:java:io:iceberg")
+  testImplementation project(path: ":sdks:java:managed")
+  testImplementation "org.apache.iceberg:iceberg-api:1.10.0"
+  testImplementation "org.apache.iceberg:iceberg-core:1.10.0"
+  testImplementation library.java.hadoop_common
+  testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+}

sdks/java/extensions/openlineage/src/main/java/org/apache/beam/sdk/extensions/openlineage/BeamOpenLineageConfig.java

@@ -0,0 +1,90 @@
+/*
+ * 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.beam.sdk.extensions.openlineage;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.openlineage.client.OpenLineageConfig;
+import io.openlineage.client.job.JobConfig;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * OpenLineage configuration for Beam pipelines, extending the generic {@link OpenLineageConfig}
+ * from openlineage-java with Beam-specific entries — the same pattern as {@code
+ * SparkOpenLineageConfig} and {@code FlinkOpenLineageConfig} in the official integrations.
+ *
+ * <p>Deserialized from {@code openlineage.yml} or {@code OPENLINEAGE__} environment variables and
+ * merged with values from {@link OpenLineagePipelineOptions}.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class BeamOpenLineageConfig extends OpenLineageConfig<BeamOpenLineageConfig> {
+
+  /** Default interval between RUNNING events, matching the Flink integration. */
+  static final int DEFAULT_TRACKING_INTERVAL_SECONDS = 60;
+
+  @JsonProperty("trackingIntervalInSeconds")
+  private @Nullable Integer trackingIntervalInSeconds;
+
+  public @Nullable Integer getTrackingIntervalInSeconds() {
+    return trackingIntervalInSeconds;
+  }
+
+  public void setTrackingIntervalInSeconds(@Nullable Integer trackingIntervalInSeconds) {
+    this.trackingIntervalInSeconds = trackingIntervalInSeconds;
+  }
+
+  @Override
+  public BeamOpenLineageConfig mergeWithNonNull(BeamOpenLineageConfig other) {
+    BeamOpenLineageConfig merged = new BeamOpenLineageConfig();
+    merged.setTransportConfig(mergePropertyWith(getTransportConfig(), other.getTransportConfig()));
+    merged.setFacetsConfig(mergePropertyWith(getFacetsConfig(), other.getFacetsConfig()));
+    merged.setDatasetConfig(mergePropertyWith(getDatasetConfig(), other.getDatasetConfig()));
+    merged.setCircuitBreaker(mergePropertyWith(getCircuitBreaker(), other.getCircuitBreaker()));
+    merged.setMetricsConfig(mergePropertyWith(getMetricsConfig(), other.getMetricsConfig()));
+    merged.setRunConfig(mergePropertyWith(getRunConfig(), other.getRunConfig()));
+    JobConfig mergedJobConfig = mergeJobConfig(getJobConfig(), other.getJobConfig());
+    if (mergedJobConfig != null) {
+      merged.setJobConfig(mergedJobConfig);
+    }
+    merged.setTrackingIntervalInSeconds(
+        mergePropertyWith(trackingIntervalInSeconds, other.getTrackingIntervalInSeconds()));
+    return merged;
+  }
+
+  /**
+   * Merges job configs field by field. {@link JobConfig#mergeWithNonNull} assumes a non-null owners
+   * map on both sides and throws otherwise, so it cannot be used directly on configs assembled from
+   * pipeline options.
+   */
+  private static @Nullable JobConfig mergeJobConfig(
+      @Nullable JobConfig base, @Nullable JobConfig overrides) {
+    if (base == null) {
+      return overrides;
+    }
+    if (overrides == null) {
+      return base;
+    }
+    JobConfig merged = new JobConfig();
+    merged.setNamespace(
+        overrides.getNamespace() != null ? overrides.getNamespace() : base.getNamespace());
+    merged.setName(overrides.getName() != null ? overrides.getName() : base.getName());
+    merged.setOwners(overrides.getOwners() != null ? overrides.getOwners() : base.getOwners());
+    merged.setTags(overrides.getTags() != null ? overrides.getTags() : base.getTags());
+    return merged;
+  }
+}

sdks/java/extensions/openlineage/src/main/java/org/apache/beam/sdk/extensions/openlineage/BeamOpenLineageConfigParser.java

@@ -0,0 +1,83 @@
+/*
+ * 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.beam.sdk.extensions.openlineage;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import io.openlineage.client.DefaultConfigPathProvider;
+import io.openlineage.client.OpenLineageClientUtils;
+import io.openlineage.client.job.JobConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Resolves the effective {@link BeamOpenLineageConfig} for a pipeline, mirroring the Spark
+ * integration's {@code ArgumentParser}: pipeline options take precedence over {@code
+ * openlineage.yml}, which takes precedence over {@code OPENLINEAGE__} environment variables.
+ */
+class BeamOpenLineageConfigParser {
+
+  private static final Logger LOG = LoggerFactory.getLogger(BeamOpenLineageConfigParser.class);
+  private static final TypeReference<BeamOpenLineageConfig> TYPE_REFERENCE =
+      new TypeReference<BeamOpenLineageConfig>() {};
+
+  private BeamOpenLineageConfigParser() {}
+
+  static BeamOpenLineageConfig parse(OpenLineagePipelineOptions options) {
+    BeamOpenLineageConfig fromEnv = loadFromEnvVars();
+    BeamOpenLineageConfig fromFile = loadFromFile();
+    BeamOpenLineageConfig fromOptions = fromOptions(options);
+    return fromEnv.mergeWith(fromFile).mergeWith(fromOptions);
+  }
+
+  private static BeamOpenLineageConfig loadFromFile() {
+    try {
+      return OpenLineageClientUtils.loadOpenLineageConfigYaml(
+          new DefaultConfigPathProvider(), TYPE_REFERENCE);
+    } catch (RuntimeException e) {
+      LOG.debug("No openlineage.yml configuration found", e);
+      return new BeamOpenLineageConfig();
+    }
+  }
+
+  private static BeamOpenLineageConfig loadFromEnvVars() {
+    try {
+      return OpenLineageClientUtils.loadOpenLineageConfigFromEnvVars(TYPE_REFERENCE);
+    } catch (RuntimeException e) {
+      LOG.debug("No OPENLINEAGE__ environment configuration found", e);
+      return new BeamOpenLineageConfig();
+    }
+  }
+
+  private static BeamOpenLineageConfig fromOptions(OpenLineagePipelineOptions options) {
+    BeamOpenLineageConfig config = new BeamOpenLineageConfig();
+    String namespace = options.getOpenLineageNamespace();
+    String jobName = options.getOpenLineageJobName();
+    if (namespace != null || jobName != null) {
+      JobConfig jobConfig = new JobConfig();
+      if (namespace != null) {
+        jobConfig.setNamespace(namespace);
+      }
+      if (jobName != null) {
+        jobConfig.setName(jobName);
+      }
+      config.setJobConfig(jobConfig);
+    }
+    config.setTrackingIntervalInSeconds(options.getOpenLineageTrackingIntervalInSeconds());
+    return config;
+  }
+}