JCSMP properties providers for new SolaceIO write connector

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

@@ -758,6 +758,7 @@ class BeamModulePlugin implements Plugin<Project> {
         // [bomupgrader] the BOM version is set by scripts/tools/bomupgrader.py. If update manually, also update
         // libraries-bom version on sdks/java/container/license_scripts/dep_urls_java.yaml
         google_cloud_platform_libraries_bom         : "com.google.cloud:libraries-bom:26.39.0",
+        google_cloud_secret_manager                 : "com.google.cloud:google-cloud-secretmanager", // google_cloud_platform_libraries_bom sets version
         google_cloud_spanner                        : "com.google.cloud:google-cloud-spanner", // google_cloud_platform_libraries_bom sets version
         google_cloud_spanner_test                   : "com.google.cloud:google-cloud-spanner:$google_cloud_spanner_version:tests",
         google_cloud_vertexai                       : "com.google.cloud:google-cloud-vertexai", // google_cloud_platform_libraries_bom sets version
@@ -858,6 +859,7 @@ class BeamModulePlugin implements Plugin<Project> {
         proto_google_cloud_firestore_v1             : "com.google.api.grpc:proto-google-cloud-firestore-v1", // google_cloud_platform_libraries_bom sets version
         proto_google_cloud_pubsub_v1                : "com.google.api.grpc:proto-google-cloud-pubsub-v1", // google_cloud_platform_libraries_bom sets version
         proto_google_cloud_pubsublite_v1            : "com.google.api.grpc:proto-google-cloud-pubsublite-v1", // google_cloud_platform_libraries_bom sets version
+        proto_google_cloud_secret_manager_v1        : "com.google.api.grpc:proto-google-cloud-secretmanager-v1", // google_cloud_platform_libraries_bom sets version
         proto_google_cloud_spanner_v1               : "com.google.api.grpc:proto-google-cloud-spanner-v1", // google_cloud_platform_libraries_bom sets version
         proto_google_cloud_spanner_admin_database_v1: "com.google.api.grpc:proto-google-cloud-spanner-admin-database-v1", // google_cloud_platform_libraries_bom sets version
         proto_google_common_protos                  : "com.google.api.grpc:proto-google-common-protos", // google_cloud_platform_libraries_bom sets version

sdks/java/io/solace/build.gradle

@@ -34,6 +34,9 @@ dependencies {
     implementation library.java.joda_time
     implementation library.java.solace
     implementation library.java.google_cloud_core
+    implementation library.java.google_cloud_secret_manager
+    implementation library.java.proto_google_cloud_secret_manager_v1
+    implementation library.java.protobuf_java
     implementation library.java.vendored_guava_32_1_2_jre
     implementation project(":sdks:java:extensions:avro")
     implementation library.java.avro

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java

@@ -181,6 +181,10 @@
  *
  * }</pre>
  *
+ * <h2>Writing</h2>
+ *
+ * TBD
+ *
  * <h3>Authentication</h3>
  *
  * <p>When reading from Solace, the user must use {@link
@@ -209,6 +213,12 @@ public class SolaceIO {
       };
   private static final boolean DEFAULT_DEDUPLICATE_RECORDS = false;
 
+  // Part of the new write connector, documentation to be updated in upcoming pull requests
+  public enum SubmissionMode {
+    HIGHER_THROUGHPUT,
+    LOWER_LATENCY
+  }
+
   /** Get a {@link Topic} object from the topic name. */
   static Topic topicFromName(String topicName) {
     return JCSMPFactory.onlyInstance().createTopic(topicName);

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/BasicAuthJcsmpSessionService.java

@@ -39,7 +39,7 @@
  * <p>This class provides a way to connect to a Solace broker and receive messages from a queue. The
  * connection is established using basic authentication.
  */
-public class BasicAuthJcsmpSessionService implements SessionService {
+public class BasicAuthJcsmpSessionService extends SessionService {
   private final String queueName;
   private final String host;
   private final String username;
@@ -137,12 +137,19 @@ private int connectSession() throws JCSMPException {
   }
 
   private JCSMPSession createSessionObject() throws InvalidPropertiesException {
-    JCSMPProperties properties = new JCSMPProperties();
-    properties.setProperty(JCSMPProperties.HOST, host);
-    properties.setProperty(JCSMPProperties.USERNAME, username);
-    properties.setProperty(JCSMPProperties.PASSWORD, password);
-    properties.setProperty(JCSMPProperties.VPN_NAME, vpnName);
-
+    JCSMPProperties properties = initializeSessionProperties(new JCSMPProperties());
     return JCSMPFactory.onlyInstance().createSession(properties);
   }
+
+  @Override
+  public JCSMPProperties initializeSessionProperties(JCSMPProperties baseProps) {
+    baseProps.setProperty(JCSMPProperties.VPN_NAME, vpnName);
+
+    baseProps.setProperty(
+        JCSMPProperties.AUTHENTICATION_SCHEME, JCSMPProperties.AUTHENTICATION_SCHEME_BASIC);
+    baseProps.setProperty(JCSMPProperties.USERNAME, username);
+    baseProps.setProperty(JCSMPProperties.PASSWORD, password);
+    baseProps.setProperty(JCSMPProperties.HOST, host);
+    return baseProps;
+  }
 }

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/BasicAuthJcsmpSessionServiceFactory.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.io.solace.broker;
 
+import static org.apache.beam.sdk.io.solace.broker.SessionService.DEFAULT_VPN_NAME;
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 
 import com.google.auto.value.AutoValue;
@@ -39,7 +40,7 @@ public abstract class BasicAuthJcsmpSessionServiceFactory extends SessionService
   public abstract String vpnName();
 
   public static Builder builder() {
-    return new AutoValue_BasicAuthJcsmpSessionServiceFactory.Builder();
+    return new AutoValue_BasicAuthJcsmpSessionServiceFactory.Builder().vpnName(DEFAULT_VPN_NAME);
   }
 
   @AutoValue.Builder

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/GCPSecretSessionServiceFactory.java

@@ -0,0 +1,169 @@
+/*
+ * 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.io.solace.broker;
+
+import static org.apache.beam.sdk.io.solace.broker.SessionService.DEFAULT_VPN_NAME;
+
+import com.google.auto.value.AutoValue;
+import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
+import com.google.cloud.secretmanager.v1.SecretVersionName;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements a {@link SessionServiceFactory} that retrieve the basic authentication
+ * credentials from a Google Cloud Secret Manager secret.
+ *
+ * <p>It can be used to avoid having to pass the password as an option of your pipeline. For this
+ * provider to work, the worker where the job runs needs to have the necessary credentials to access
+ * the secret. In Dataflow, this implies adding the necessary permissions to the worker service
+ * account. For other runners, set the credentials in the pipeline options using {@link
+ * org.apache.beam.sdk.extensions.gcp.options.GcpOptions}.
+ *
+ * <p>It also shows how to implement a {@link SessionServiceFactory} that depends on using external
+ * resources to retrieve the Solace session properties. In this case, using the Google Cloud Secrete
+ * Manager client.
+ *
+ * <p>Example of how to create the provider object:
+ *
+ * <pre>{@code
+ * GCPSecretSessionServiceFactory factory =
+ *     GCPSecretSessionServiceFactory.builder()
+ *         .username("user")
+ *         .host("host:port")
+ *         .passwordSecretName("secret-name")
+ *         .build();
+ *
+ * SessionService serviceUsingGCPSecret = factory.create();
+ * }</pre>
+ */
+@AutoValue
+public abstract class GCPSecretSessionServiceFactory extends SessionServiceFactory {
+
+  private static final Logger LOG = LoggerFactory.getLogger(GCPSecretSessionServiceFactory.class);
+
+  private static final String PROJECT_NOT_FOUND = "PROJECT-NOT-FOUND";
+
+  public abstract String username();
+
+  public abstract String host();
+
+  public abstract String passwordSecretName();
+
+  public abstract String vpnName();
+
+  public abstract @Nullable String secretManagerProjectId();
+
+  public abstract String passwordSecretVersion();
+
+  public static GCPSecretSessionServiceFactory.Builder builder() {
+    return new AutoValue_GCPSecretSessionServiceFactory.Builder()
+        .passwordSecretVersion("latest")
+        .vpnName(DEFAULT_VPN_NAME);
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    /** Username to be used to authenticate with the broker. */
+    public abstract GCPSecretSessionServiceFactory.Builder username(String username);
+
+    /**
+     * The location of the broker, including port details if it is not listening in the default
+     * port.
+     */
+    public abstract GCPSecretSessionServiceFactory.Builder host(String host);
+
+    /** The Secret Manager secret name where the password is stored. */
+    public abstract GCPSecretSessionServiceFactory.Builder passwordSecretName(String name);
+
+    /** Optional. Solace broker VPN name. If not set, "default" is used. */
+    public abstract GCPSecretSessionServiceFactory.Builder vpnName(String name);
+
+    /**
+     * Optional for Dataflow or VMs running on Google Cloud. The project id of the project where the
+     * secret is stored. If not set, the project id where the job is running is used.
+     */
+    public abstract GCPSecretSessionServiceFactory.Builder secretManagerProjectId(String id);
+
+    /** Optional. Solace broker password secret version. If not set, "latest" is used. */
+    public abstract GCPSecretSessionServiceFactory.Builder passwordSecretVersion(String version);
+
+    public abstract GCPSecretSessionServiceFactory build();
+  }
+
+  @Override
+  public SessionService create() {
+    String password = null;
+    try {
+      password = retrieveSecret();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+
+    BasicAuthJcsmpSessionServiceFactory factory =
+        BasicAuthJcsmpSessionServiceFactory.builder()
+            .username(username())
+            .host(host())
+            .password(password)
+            .vpnName(vpnName())
+            .build();
+
+    return factory.create();
+  }
+
+  private String retrieveSecret() throws IOException {
+    try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) {
+      String projectId =
+          Optional.ofNullable(secretManagerProjectId()).orElse(getProjectIdFromVmMetadata());
+      SecretVersionName secretVersionName =
+          SecretVersionName.of(projectId, passwordSecretName(), passwordSecretVersion());
+      return client.accessSecretVersion(secretVersionName).getPayload().getData().toStringUtf8();
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private String getProjectIdFromVmMetadata() throws IOException {
+    URL metadataUrl =
+        new URL("http://metadata.google.internal/computeMetadata/v1/project/project-id");
+    HttpURLConnection connection = (HttpURLConnection) metadataUrl.openConnection();
+    connection.setRequestProperty("Metadata-Flavor", "Google");
+
+    String output;
+    try (BufferedReader reader =
+        new BufferedReader(
+            new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) {
+      output = reader.readLine();
+    }
+
+    if (output == null || output.isEmpty()) {
+      LOG.error(
+          "Cannot retrieve project id from VM metadata, please set a project id in your GoogleCloudSecretProvider.");
+    }
+    return output != null ? output : PROJECT_NOT_FOUND;
+  }
+}