[Java] Support dynamic secret provider registration via SecretRegistrar

sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java

@@ -0,0 +1,34 @@
+/*
+ * 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.util;
+
+import com.google.auto.service.AutoService;
+import java.util.Map;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+
+/** {@link AutoService} registrar for the {@link GcpHsmGeneratedSecret}. */
+@AutoService(SecretRegistrar.class)
+public class GcpHsmGeneratedSecretRegistrar implements SecretRegistrar {
+
+  @Override
+  public Map<String, SecretFactory> getSecretFactories() {
+    return ImmutableMap.of(
+        "googlecloudhsmgeneratedsecretmanager", GcpHsmGeneratedSecret::fromMap,
+        "gcphsmgeneratedsecret", GcpHsmGeneratedSecret::fromMap);
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java

@@ -0,0 +1,34 @@
+/*
+ * 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.util;
+
+import com.google.auto.service.AutoService;
+import java.util.Map;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+
+/** {@link AutoService} registrar for the {@link GcpSecret}. */
+@AutoService(SecretRegistrar.class)
+public class GcpSecretRegistrar implements SecretRegistrar {
+
+  @Override
+  public Map<String, SecretFactory> getSecretFactories() {
+    return ImmutableMap.of(
+        "googlecloudsecretmanager", GcpSecret::fromMap,
+        "gcpsecret", GcpSecret::fromMap);
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/util/Secret.java

@@ -21,8 +21,11 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.beam.sdk.util.common.ReflectHelpers;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,6 +38,30 @@
  * should be able to return a valid byte array representing the secret.
  */
 public abstract class Secret implements Serializable {
+  private static final Logger LOG = LoggerFactory.getLogger(Secret.class);
+  private static final Map<String, SecretRegistrar.SecretFactory> SECRET_FACTORIES =
+      loadSecretFactories();
+
+  private static Map<String, SecretRegistrar.SecretFactory> loadSecretFactories() {
+    Map<String, SecretRegistrar.SecretFactory> factories = new HashMap<>();
+    for (SecretRegistrar registrar : ReflectHelpers.loadServicesOrdered(SecretRegistrar.class)) {
+      for (Map.Entry<String, SecretRegistrar.SecretFactory> entry :
+          registrar.getSecretFactories().entrySet()) {
+        String key = entry.getKey().toLowerCase();
+        if (factories.containsKey(key)) {
+          throw new IllegalStateException(
+              String.format(
+                  "Duplicate SecretRegistrar for secret manager name '%s': %s and %s",
+                  key,
+                  factories.get(key).getClass().getName(),
+                  entry.getValue().getClass().getName()));
+        }
+        factories.put(key, entry.getValue());
+      }
+    }
+    return ImmutableMap.copyOf(factories);
+  }
+
   private transient byte @Nullable [] cachedSecretBytes = null;
 
   /**
@@ -104,29 +131,23 @@ public static Secret parseSecretOption(String secretOption) {
     }
 
     String secretType = rawType.toLowerCase();
-    String secretManager;
-    switch (secretType) {
-      case "gcpsecret":
-        secretManager = "GoogleCloudSecretManager";
-        break;
-      case "gcphsmgeneratedsecret":
-        secretManager = "GoogleCloudHsmGeneratedSecretManager";
-        break;
-      default:
-        throw new IllegalArgumentException(
-            String.format(
-                "Invalid secret type %s, currently only GcpSecret and GcpHsmGeneratedSecret are supported",
-                secretType));
+    SecretRegistrar.SecretFactory factory = SECRET_FACTORIES.get(secretType);
+    if (factory == null) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Invalid secret type %s, currently supported types: %s",
+              rawType, SECRET_FACTORIES.keySet()));
     }
 
     try {
-      ObjectMapper mapper = new ObjectMapper();
-      String jsonSpec = mapper.writeValueAsString(paramMap);
-      return fromJson(jsonSpec, secretManager);
+      return factory.createSecret(paramMap);
     } catch (Exception e) {
       if (e instanceof IllegalArgumentException) {
         throw (IllegalArgumentException) e;
       }
+      if (e instanceof NullPointerException) {
+        throw (NullPointerException) e;
+      }
       throw new RuntimeException("Failed to parse secret option", e);
     }
   }
@@ -139,7 +160,6 @@ public static Secret parseSecretOption(String secretOption) {
    * @return An instance of Secret.
    */
   public static Secret fromJson(@Nullable String spec, @Nullable String secretManager) {
-    Logger logger = LoggerFactory.getLogger(Secret.class);
     String smManager = secretManager != null ? secretManager.trim() : null;
     if (smManager != null && smManager.isEmpty()) {
       smManager = null;
@@ -152,38 +172,23 @@ public static Secret fromJson(@Nullable String spec, @Nullable String secretMana
         mapper.configure(com.fasterxml.jackson.core.JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
         specMap = mapper.readValue(spec, new TypeReference<Map<String, String>>() {});
       } catch (Exception e) {
-        logger.debug("Failed to parse secret spec as JSON map", e);
+        LOG.debug("Failed to parse secret spec as JSON map", e);
       }
     }
 
     if (smManager != null) {
-      switch (smManager.toLowerCase()) {
-        case "googlecloudsecretmanager":
-        case "gcpsecret":
-          if (specMap != null) {
-            return GcpSecret.fromMap(specMap);
-          } else if (spec != null) {
-            return new GcpSecret(spec);
-          } else {
-            throw new IllegalArgumentException("Invalid spec for GcpSecret");
-          }
-        case "googlecloudhsmgeneratedsecretmanager":
-        case "gcphsmgeneratedsecret":
-          if (specMap != null) {
-            return GcpHsmGeneratedSecret.fromMap(specMap);
-          } else {
-            throw new IllegalArgumentException("Invalid spec for GcpHsmGeneratedSecret");
-          }
-        default:
-          throw new IllegalArgumentException(
-              String.format(
-                  "Unsupported secret manager: '%s'. Currently supported options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'.",
-                  smManager));
+      SecretRegistrar.SecretFactory factory = SECRET_FACTORIES.get(smManager.toLowerCase());
+      if (factory != null) {
+        return factory.createSecret(specMap != null ? specMap : Collections.emptyMap());
       }
+      throw new IllegalArgumentException(
+          String.format(
+              "Unsupported secret manager: '%s'. Currently supported options: %s.",
+              smManager, SECRET_FACTORIES.keySet()));
     }
 
     if (specMap != null) {
-      logger.warn(
+      LOG.warn(
           "The 'spec' parameter appears to be a JSON specification, but 'secret_manager' is not set. Defaulting to Raw.");
     }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/util/SecretRegistrar.java

@@ -0,0 +1,52 @@
+/*
+ * 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.util;
+
+import com.google.auto.service.AutoService;
+import java.util.Map;
+import java.util.ServiceLoader;
+
+/**
+ * A registrar that creates {@link Secret} instances from a spec parameter map.
+ *
+ * <p>{@link Secret} creators have the ability to provide a registrar by creating a {@link
+ * ServiceLoader} entry and a concrete implementation of this interface.
+ *
+ * <p>It is optional but recommended to use one of the many build time tools such as {@link
+ * AutoService} to generate the necessary META-INF files automatically.
+ */
+public interface SecretRegistrar {
+
+  /** Functional interface for creating a {@link Secret} from a specification map. */
+  @FunctionalInterface
+  interface SecretFactory {
+    /**
+     * Creates a {@link Secret} instance from a spec parameter map.
+     *
+     * @param specMap The parsed map of key-value parameters.
+     * @return The constructed {@link Secret} instance.
+     */
+    Secret createSecret(Map<String, String> specMap);
+  }
+
+  /**
+   * Returns a map from secret provider name / type (case-insensitive) to the corresponding {@link
+   * SecretFactory}.
+   */
+  Map<String, SecretFactory> getSecretFactories();
+}

sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.util;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItems;
+import static org.junit.Assert.fail;
+
+import java.util.Map;
+import java.util.ServiceLoader;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GcpSecretRegistrar} and {@link GcpHsmGeneratedSecretRegistrar}. */
+@RunWith(JUnit4.class)
+public class GcpSecretRegistrarTest {
+
+  @Test
+  public void testGcpSecretRegistrarServiceLoader() {
+    for (SecretRegistrar registrar :
+        Lists.newArrayList(ServiceLoader.load(SecretRegistrar.class).iterator())) {
+      if (registrar instanceof GcpSecretRegistrar) {
+        Map<String, SecretRegistrar.SecretFactory> factories = registrar.getSecretFactories();
+        assertThat(factories.keySet(), hasItems("googlecloudsecretmanager", "gcpsecret"));
+        return;
+      }
+    }
+    fail("Expected to find " + GcpSecretRegistrar.class);
+  }
+
+  @Test
+  public void testGcpHsmGeneratedSecretRegistrarServiceLoader() {
+    for (SecretRegistrar registrar :
+        Lists.newArrayList(ServiceLoader.load(SecretRegistrar.class).iterator())) {
+      if (registrar instanceof GcpHsmGeneratedSecretRegistrar) {
+        Map<String, SecretRegistrar.SecretFactory> factories = registrar.getSecretFactories();
+        assertThat(
+            factories.keySet(),
+            hasItems("googlecloudhsmgeneratedsecretmanager", "gcphsmgeneratedsecret"));
+        return;
+      }
+    }
+    fail("Expected to find " + GcpHsmGeneratedSecretRegistrar.class);
+  }
+}