Added implementation for openFile and openFileWithOptions

Author: prince-cs

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).
@@ -103,6 +109,57 @@ public FSDataInputStream open(Path path, int bufferSize) throws IOException {
     return new FSDataInputStream(new SeekableByteChannelFSInputStream(decryptor.open(fs, path, bufferSize)));
   }
 
+  @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, 4096)));
+          } catch (Exception e) {
+            throw new CompletionException(e);
+          }
+        });
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder opt(@NotNull String s, @NotNull String s1) {
+        return null;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder opt(@NotNull String s, @NotNull String... strings) {
+        return null;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder must(@NotNull String s, @NotNull String s1) {
+        return null;
+      }
+
+      @Override
+      public FutureDataInputStreamBuilder must(@NotNull String s, @NotNull String... strings) {
+        return null;
+      }
+    };
+  }
+
+  @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}.
    */