Update options

Author: clun

release.properties

@@ -0,0 +1,11 @@
+#release configuration
+#Wed Jun 21 19:46:01 CEST 2023
+projectVersionPolicyId=default
+scm.tagNameFormat=@{project.version}
+remoteTagging=true
+scm.commentPrefix=[maven-release-plugin] 
+pushChanges=true
+completedPhase=check-poms
+scm.url=scm\:git\:git@github.com\:clun/beam-sdks-java-io-astra.git
+exec.snapshotReleasePluginAllowed=false
+preparationGoals=clean verify

src/main/java/org/apache/beam/sdk/io/astra/db/AstraDbIO.java

@@ -853,4 +853,8 @@ public PCollection<T> expand(PCollection<Read<T>> input) {
     }
   }
 
+  public static void close() {
+    CqlSessionHolder.cleanup();
+  }
+
 }

src/main/java/org/apache/beam/sdk/io/astra/db/CqlSessionHolder.java

@@ -168,14 +168,14 @@ private static synchronized void init() {
         throw new IllegalStateException("SHA-1 is not supported");
       }
       // Add a Shutdown Hook to close all sessions.
-      Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-        LOG.debug("CqlSessionHolder closing sessions...");
-        cacheSessions.values().stream()
-                .filter(s->!s.isClosed())
-                .forEach(CqlSession::close);
-        LOG.debug("CqlSessionHolder sessions are closed.");
-      }));
+      Runtime.getRuntime().addShutdownHook(new Thread(() -> { cleanup(); }));
     }
   }
 
+  public static void cleanup() {
+    cacheSessions.values().stream()
+            .filter(s->!s.isClosed())
+            .forEach(CqlSession::close);
+  }
+
 }

src/main/java/org/apache/beam/sdk/io/astra/db/ExecuteCqlFn.java

@@ -22,8 +22,6 @@
 
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.cql.Row;
-import org.apache.beam.sdk.io.astra.db.AstraDbIO;
-import org.apache.beam.sdk.io.astra.db.CqlSessionHolder;
 import org.apache.beam.sdk.transforms.DoFn;
 
 /**

src/main/java/org/apache/beam/sdk/io/astra/db/ReadFn.java

@@ -58,7 +58,6 @@
 
 import java.math.BigInteger;
 import java.util.Iterator;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 /**

src/main/java/org/apache/beam/sdk/io/astra/db/options/AstraDbDeleteOptions.java

@@ -23,9 +23,4 @@
 /**
  * Specific keys for write operations.
  */
-public interface AstraDbDeleteOptions extends AstraDbOptions {
-
-    // no consistency level as forced to LOCAL_QUORUM
-
-    // no read timeout
-}
+public interface AstraDbDeleteOptions extends AstraDbOptions {}

src/main/java/org/apache/beam/sdk/io/astra/db/options/AstraDbOptions.java

@@ -9,9 +9,9 @@
  * Licensed under the Apache License, Version 2.0
  * 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.
@@ -46,36 +46,41 @@ public interface AstraDbOptions extends PipelineOptions  {
     void setAstraToken(String token);
 
     /**
-     * Access Astra secure bundle
+     * Secure Connect Bundle location. It can be;
+     * - a secret resourceId
+     * - a path to a file
+     * - a URL
+     *
+     * It will be converted as a byte array for Astra DbIO
      * @return the Astra secure bundle
      */
     @Description("Location of secure connect bundle, depending on environment could be path or secret resource id")
     @Validation.Required
-    byte[] getAstraSecureConnectBundle();
+    String getAstraSecureConnectBundle();
 
     /**
      * Update the Astra secure bundle
      *
-     * @param path
+     * @param secureConnectBundleLocation
      *      new value for Astra connection timeout
      */
     @SuppressWarnings("not used")
-    void setAstraSecureConnectBundle(byte[] path);
+    void setAstraSecureConnectBundle(String secureConnectBundleLocation);
 
     /**
      * Access Astra Keyspace
      * @return the Astra keyspace
      */
     @Description("Keyspace in Cassandra, a Db can have multiple keyspace")
     @Validation.Required
-    String getKeyspace();
+    String getAstraKeyspace();
 
     /**
      * Update the Astra keyspace
      *
      * @param keyspace
      *      new value for Astra keyspace
      */
-    void setKeyspace(String keyspace);
+    void setAstraKeyspace(String keyspace);
 
 }

src/main/java/org/apache/beam/sdk/io/astra/db/options/AstraDbWriteOptions.java

@@ -23,9 +23,4 @@
 /**
  * Specific keys for write operations.
  */
-public interface AstraDbWriteOptions extends AstraDbOptions {
-
-    // no consistency level as forced to LOCAL_QUORUM
-
-    // no read timeout
-}
+public interface AstraDbWriteOptions extends AstraDbOptions {}

src/main/java/org/apache/beam/sdk/io/astra/db/transforms/RunCqlQueryFn.java

@@ -20,9 +20,7 @@
  * #L%
  */
 
-import org.apache.beam.sdk.io.astra.db.AstraDbIO;
 import org.apache.beam.sdk.io.astra.db.CqlSessionHolder;
-import org.apache.beam.sdk.io.astra.db.options.AstraDbOptions;
 import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.PCollection;
@@ -42,18 +40,6 @@ public class RunCqlQueryFn<T> extends PTransform<PCollection<T>, PCollection<T>>
      */
     private static final Logger LOG = LoggerFactory.getLogger(CqlSessionHolder.class);
 
-    /**
-     * Execute a CQL query
-     *
-     * @param options
-     *         pipeline Options
-     * @param cql
-     *         cql command to execute
-     */
-    public RunCqlQueryFn(AstraDbOptions options, String cql) {
-        this(options.getAstraToken(), options.getAstraSecureConnectBundle(), options.getKeyspace(), cql);
-    }
-
     /**
      * Execute a CQL query
      *

src/main/java/org/apache/beam/sdk/io/astra/db/utils/AstraSecureConnectBundleUtils.java

@@ -0,0 +1,107 @@
+package org.apache.beam.sdk.io.astra.db.utils;
+
+/*-
+ * #%L
+ * Beam SDK for Astra
+ * --
+ * Copyright (C) 2023 DataStax
+ * --
+ * Licensed under the Apache License, Version 2.0
+ * 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.
+ * #L%
+ */
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Utility class for Astra Secure Connect Bundle.
+ */
+public class AstraSecureConnectBundleUtils {
+
+    /**
+     * Private constructor.
+     */
+    private AstraSecureConnectBundleUtils() {}
+
+    /**
+     * Read binary content from a file.
+     *
+     * @param filePath
+     *      file Path
+     * @return
+     *      binary
+     */
+    public static final byte[] loadFromFilePath(String filePath) {
+        try {
+            return Files.readAllBytes(Path.of(filePath));
+        } catch(Exception e) {
+            throw new RuntimeException("Unable to read file " + filePath, e);
+        }
+    }
+
+    /**
+     * Download a file and get binary from URL.
+     *
+     * @param fileUrl
+     *      file URL
+     * @return
+     *      binary content
+     */
+    public static final byte[] loadFromURL(String fileUrl) {
+        URL url = null;
+        try {
+            url = new URL(fileUrl);
+            try (
+                 InputStream inputStream = url.openStream();
+                 ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
+                byte[] buffer = new byte[4096];
+                int bytesRead;
+                while ((bytesRead = inputStream.read(buffer)) != -1) {
+                    outputStream.write(buffer, 0, bytesRead);
+                }
+                return outputStream.toByteArray();
+            }
+        } catch (MalformedURLException e) {
+            throw new RuntimeException("Invalid URL " + fileUrl, e);
+        } catch (IOException e) {
+            throw new RuntimeException("Unable to read file " + fileUrl, e);
+        }
+    }
+
+    /**
+     * Get binary content from an inputStream.
+     *
+     * @param inputStream
+     *      inputstream
+     * @return
+     *     binary content   
+     */
+    public static final byte[] loadFromInputStream(InputStream inputStream) {
+        try {
+            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+            byte[] buffer = new byte[4096];
+            int bytesRead;
+            while ((bytesRead = inputStream.read(buffer)) != -1) {
+                outputStream.write(buffer, 0, bytesRead);
+            }
+            return outputStream.toByteArray();
+        } catch (IOException e) {
+            throw new RuntimeException("Unable to read input stream", e);
+        }
+    }
+}