Added implementation for new openFile for tink enc

Author: psainics

pom.xml

@@ -89,7 +89,7 @@
     <google.tink.version>1.3.0-rc3</google.tink.version>
     <guava.version>27.0.1-jre</guava.version>
     <hadoop.version>3.3.6</hadoop.version>
-    <hbase-shaded-client.version>1.4.13</hbase-shaded-client.version>
+    <hbase-shaded-client.version>2.6.2-hadoop3</hbase-shaded-client.version>
     <hbase-shaded-server.version>1.4.13</hbase-shaded-server.version>
     <httpclient.version>4.5.13</httpclient.version>
     <jackson.core.version>2.13.4.2</jackson.core.version>

src/main/java/io/cdap/plugin/gcp/crypto/EncryptedFileSystem.java

@@ -22,7 +22,11 @@
 import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathHandle;
+import org.apache.hadoop.fs.impl.OpenFileParameters;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,6 +36,8 @@
 import java.nio.channels.Channels;
 import java.nio.channels.SeekableByteChannel;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 
 /**
  * A hadoop {@link FileSystem} that support files decryption (encryption is currently not supported).
@@ -42,6 +48,7 @@ public class EncryptedFileSystem extends FilterFileSystem {
   private static final String FS_SCHEME = CONF_PREFIX + "scheme";
   private static final String FS_IMPL = CONF_PREFIX + "impl";
   private static final String DECRYPTOR_IMPL = CONF_PREFIX + "decryptor.impl";
+  private static final int DEFAULT_BUFFER_SIZE = 4096;
 
   private static final Logger LOG = LoggerFactory.getLogger(EncryptedFileSystem.class);
 
@@ -103,6 +110,82 @@ public FSDataInputStream open(Path path, int bufferSize) throws IOException {
     return new FSDataInputStream(new SeekableByteChannelFSInputStream(decryptor.open(fs, path, bufferSize)));
   }
 
+  /**
+   * Opens a file asynchronously and returns a {@link FutureDataInputStreamBuilder}
+   * to build a {@link FSDataInputStream} for the specified {@link Path}.
+   *
+   * <p>This implementation returns a builder that constructs an input stream by using a decryptor
+   * to open the file through a {@link SeekableByteChannelFSInputStream}. The file is read
+   * with a buffer size of 4096 bytes.</p>
+   *
+   * @param path the {@link Path} of the file to open
+   * @return a {@link FutureDataInputStreamBuilder} that asynchronously builds a {@link FSDataInputStream}
+   * @throws UnsupportedOperationException if the operation is not supported
+   */
+  @Override
+  public FutureDataInputStreamBuilder openFile(Path path) throws UnsupportedOperationException {
+    return new FutureDataInputStreamBuilder() {
+      @Override
+      public CompletableFuture<FSDataInputStream> build()
+        throws IllegalArgumentException, UnsupportedOperationException {
+        return CompletableFuture.supplyAsync(() -> {
+          try {
+            return new FSDataInputStream(
+              new SeekableByteChannelFSInputStream(decryptor.open(fs, path, DEFAULT_BUFFER_SIZE)));
+          } catch (Exception e) {
+            throw new CompletionException(e);
+          }
+        });
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder opt(@NotNull String s, @NotNull String s1) {
+        return this;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder opt(@NotNull String s, @NotNull String... strings) {
+        return this;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder must(@NotNull String s, @NotNull String s1) {
+        return this;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder must(@NotNull String s, @NotNull String... strings) {
+        return this;
+      }
+    };
+  }
+
+  /**
+   * Opens a file asynchronously using the provided {@link Path}, and returns
+   * a {@link CompletableFuture} that supplies a {@link FSDataInputStream}.
+   *
+   * <p>This method uses a decryptor to open the file and wraps it in a {@link SeekableByteChannelFSInputStream}.
+   * It uses the buffer size specified in the {@code parameters}; if the buffer size is not greater than zero,
+   * a default of 4096 bytes is used.</p>
+   *
+   * @param path the {@link Path} to the file to open
+   * @param parameters the {@link OpenFileParameters} containing optional configuration, such as buffer size
+   * @return a {@link CompletableFuture} that will complete with the {@link FSDataInputStream}
+   * @throws CompletionException if an exception occurs during file opening
+   */
+  @Override
+  protected CompletableFuture<FSDataInputStream> openFileWithOptions(Path path, OpenFileParameters parameters) {
+    return CompletableFuture.supplyAsync(() -> {
+      try {
+        int bufferSize = parameters.getBufferSize() > 0 ? parameters.getBufferSize() : 4096;
+        return new FSDataInputStream(
+          new SeekableByteChannelFSInputStream(decryptor.open(fs, path, bufferSize)));
+      } catch (Exception e) {
+        throw new CompletionException(e);
+      }
+    });
+  }
+
   /**
    * A {@link FSInputStream} implementation backed by a {@link SeekableByteChannel}.
    */