LISAMZA-36419: support offspring

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -398,7 +398,7 @@ class BeamModulePlugin implements Plugin<Project> {
 
     // Automatically use the official release version if we are performing a release
     // otherwise append '-SNAPSHOT'
-    project.version = '2.45.16'
+    project.version = '2.45.17'
     if (isLinkedin(project)) {
       project.ext.mavenGroupId = 'com.linkedin.beam'
     }

gradle.properties

@@ -30,8 +30,8 @@ signing.gnupg.useLegacyGpg=true
 # buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy.
 # To build a custom Beam version make sure you change it in both places, see
 # https://github.com/apache/beam/issues/21302.
-version=2.45.16
-sdk_version=2.45.16
+version=2.45.17
+sdk_version=2.45.17
 
 javaVersion=1.8
 

sdks/java/core/src/main/java/org/apache/beam/sdk/options/CustomPipelineOptionsInitializer.java

@@ -0,0 +1,43 @@
+package org.apache.beam.sdk.options;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterators;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+
+/**
+ * Interface to support offspring wire-in for Li: if input class is a
+ * customized pipelineOptions created by an offspring factory,
+ * the pipelineOptions will be initialized.
+ *
+ */
+@SuppressWarnings("rawtypes")
+public interface CustomPipelineOptionsInitializer<T> {
+  void init(T pipelineOptions, Class<T> clazz);
+
+  /**
+   * Inject the implementation for the interface
+   * <p>Usage:
+   *
+   * <pre>{@code
+   *  @AutoService(CustomPipelineOptionsInitializer.Registrar.class)
+   *   public static class Registrar implements CustomPipelineOptions.Registrar {
+   *     @Override
+   *     public CustomPipelineOptions create() {
+   *       return new CustomPipelineOptions();
+   *     }
+   *   }</pre>
+   */
+  interface Registrar {
+    CustomPipelineOptionsInitializer create();
+  }
+
+
+  static @Initialized @Nullable CustomPipelineOptionsInitializer get() {
+    final Iterator<CustomPipelineOptionsInitializer.Registrar>
+        initializer = ServiceLoader.load(CustomPipelineOptionsInitializer.Registrar.class).iterator();
+    return initializer.hasNext() ? Iterators.getOnlyElement(initializer).create() : null;
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java

@@ -196,7 +196,7 @@ public Object invoke(Object proxy, Method method, Object[] args) {
     } else if (args != null && "as".equals(method.getName()) && args[0] instanceof Class) {
       @SuppressWarnings("unchecked")
       Class<? extends PipelineOptions> clazz = (Class<? extends PipelineOptions>) args[0];
-      return as(clazz);
+      return as(clazz, (PipelineOptions)proxy); // LI-specific change to wire in offspring
     } else if (args != null
         && "populateDisplayData".equals(method.getName())
         && args[0] instanceof DisplayData.Builder) {
@@ -275,6 +275,10 @@ static BoundValue fromDefault(@Nullable Object value) {
    * @return An object that implements the interface {@code <T>}.
    */
   <T extends PipelineOptions> T as(Class<T> iface) {
+    return as(iface, null);
+  }
+
+  <T extends PipelineOptions> T as(Class<T> iface, PipelineOptions pipelineOptions) {
     checkNotNull(iface);
     checkArgument(iface.isInterface(), "Not an interface: %s", iface);
 
@@ -300,6 +304,10 @@ <T extends PipelineOptions> T as(Class<T> iface) {
 
           computedProperties =
               computedProperties.updated(iface, existingOption, propertyDescriptors);
+          // Linkedin specific change: initialize the offspring generator
+          if (pipelineOptions != null && CustomPipelineOptionsInitializer.get() != null) {
+            CustomPipelineOptionsInitializer.get().init(existingOption, iface);
+          }
         }
       }
     }