HDDS-10338. Implement a Client Datanode API to stream a block

hadoop-hdds/client/dev-support/findbugsExcludeFile.xml

@@ -32,4 +32,8 @@
     <Class name="org.apache.hadoop.hdds.scm.storage.TestBlockInputStream"></Class>
     <Bug pattern="RR_NOT_CHECKED" />
   </Match>
+  <Match>
+    <Class name="org.apache.hadoop.hdds.scm.storage.TestStreamBlockInput"></Class>
+    <Bug pattern="RR_NOT_CHECKED" />
+  </Match>
 </FindBugsFilter>

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java

@@ -128,6 +128,13 @@ public enum ChecksumCombineMode {
       tags = ConfigTag.CLIENT)
   private long streamBufferMaxSize = 32 * 1024 * 1024;
 
+  @Config(key = "stream.readblock.enable",
+      defaultValue = "false",
+      type = ConfigType.BOOLEAN,
+      description = "Allow ReadBlock to stream all the readChunk in one request.",
+      tags = ConfigTag.CLIENT)
+  private boolean streamReadBlock = true;
+
   @Config(key = "max.retries",
       defaultValue = "5",
       description = "Maximum number of retries by Ozone Client on "
@@ -166,7 +173,7 @@ public enum ChecksumCombineMode {
       description = "The checksum type [NONE/ CRC32/ CRC32C/ SHA256/ MD5] "
           + "determines which algorithm would be used to compute checksum for "
           + "chunk data. Default checksum type is CRC32.",
-      tags = { ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE })
+      tags = {ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE})
   private String checksumType = ChecksumType.CRC32.name();
 
   @Config(key = "bytes.per.checksum",
@@ -175,7 +182,7 @@ public enum ChecksumCombineMode {
       description = "Checksum will be computed for every bytes per checksum "
           + "number of bytes and stored sequentially. The minimum value for "
           + "this config is 8KB.",
-      tags = { ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE })
+      tags = {ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE})
   private int bytesPerChecksum = 16 * 1024;
 
   @Config(key = "verify.checksum",
@@ -558,4 +565,12 @@ public void setMaxConcurrentWritePerKey(int maxConcurrentWritePerKey) {
   public int getMaxConcurrentWritePerKey() {
     return this.maxConcurrentWritePerKey;
   }
+
+  public boolean isStreamReadBlock() {
+    return streamReadBlock;
+  }
+
+  public void setStreamReadBlock(boolean streamReadBlock) {
+    this.streamReadBlock = streamReadBlock;
+  }
 }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java

@@ -41,6 +41,8 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
 import org.apache.hadoop.hdds.protocol.datanode.proto.XceiverClientProtocolServiceGrpc;
 import org.apache.hadoop.hdds.protocol.datanode.proto.XceiverClientProtocolServiceGrpc.XceiverClientProtocolServiceStub;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -108,12 +110,12 @@ public class XceiverClientGrpc extends XceiverClientSpi {
    * Constructs a client that can communicate with the Container framework on
    * data nodes via DatanodeClientProtocol.
    *
-   * @param pipeline - Pipeline that defines the machines.
-   * @param config   -- Ozone Config
+   * @param pipeline     - Pipeline that defines the machines.
+   * @param config       -- Ozone Config
    * @param trustManager - a {@link ClientTrustManager} with proper CA handling.
    */
   public XceiverClientGrpc(Pipeline pipeline, ConfigurationSource config,
-      ClientTrustManager trustManager) {
+                           ClientTrustManager trustManager) {
     super();
     Preconditions.checkNotNull(pipeline);
     Preconditions.checkNotNull(config);
@@ -441,7 +443,11 @@ private XceiverClientReply sendCommandWithRetry(
         // sendCommandAsyncCall will create a new channel and async stub
         // in case these don't exist for the specific datanode.
         reply.addDatanode(dn);
-        responseProto = sendCommandAsync(request, dn).getResponse().get();
+        if (request.getCmdType() == ContainerProtos.Type.ReadBlock) {
+          responseProto = sendCommandAsyncReadOnly(request, dn).getResponse().get();
+        } else {
+          responseProto = sendCommandAsync(request, dn).getResponse().get();
+        }
         if (validators != null && !validators.isEmpty()) {
           for (Validator validator : validators) {
             validator.accept(request, responseProto);
@@ -485,7 +491,7 @@ private XceiverClientReply sendCommandWithRetry(
       String message = "Failed to execute command {}";
       if (LOG.isDebugEnabled()) {
         LOG.debug(message + " on the pipeline {}.",
-                processForDebug(request), pipeline);
+            processForDebug(request), pipeline);
       } else {
         LOG.warn(message + " on the pipeline {}.",
                 request.getCmdType(), pipeline);
@@ -618,6 +624,69 @@ private void decreasePendingMetricsAndReleaseSemaphore() {
     return new XceiverClientReply(replyFuture);
   }
 
+  public XceiverClientReply sendCommandAsyncReadOnly(
+      ContainerCommandRequestProto request, DatanodeDetails dn)
+      throws IOException, InterruptedException {
+
+    CompletableFuture<ContainerCommandResponseProto> future =
+        new CompletableFuture<>();
+    ContainerCommandResponseProto.Builder response =
+        ContainerCommandResponseProto.newBuilder();
+    ContainerProtos.ReadBlockResponseProto.Builder readBlock =
+        ContainerProtos.ReadBlockResponseProto.newBuilder();
+    checkOpen(dn);
+    UUID dnID = dn.getUuid();
+    Type cmdType = request.getCmdType();
+    semaphore.acquire();
+    long requestTime = System.currentTimeMillis();
+    metrics.incrPendingContainerOpsMetrics(cmdType);
+
+    final StreamObserver<ContainerCommandRequestProto> requestObserver =
+        asyncStubs.get(dnID).withDeadlineAfter(timeout, TimeUnit.SECONDS)
+            .send(new StreamObserver<ContainerCommandResponseProto>() {
+              @Override
+              public void onNext(
+                  ContainerCommandResponseProto responseProto) {
+                if (responseProto.getResult() == Result.SUCCESS) {
+                  readBlock.addReadChunk(responseProto.getReadChunk());
+                } else {
+                  future.complete(
+                      ContainerCommandResponseProto.newBuilder(responseProto)
+                          .setCmdType(Type.StreamRead).build());
+                }
+              }
+
+              @Override
+              public void onError(Throwable t) {
+                future.completeExceptionally(t);
+                metrics.decrPendingContainerOpsMetrics(cmdType);
+                metrics.addContainerOpsLatency(
+                    cmdType, System.currentTimeMillis() - requestTime);
+
+              }
+
+              @Override
+              public void onCompleted() {
+                if (readBlock.getReadChunkCount() > 0) {
+                  future.complete(response.setReadBlock(readBlock)
+                      .setCmdType(Type.StreamRead).setResult(Result.SUCCESS).build());
+                }
+                if (!future.isDone()) {
+                  future.completeExceptionally(new IOException(
+                      "Stream completed but no reply for request " +
+                          processForDebug(request)));
+                }
+                metrics.decrPendingContainerOpsMetrics(cmdType);
+                metrics.addContainerOpsLatency(
+                    cmdType, System.currentTimeMillis() - requestTime);
+              }
+            });
+    requestObserver.onNext(request);
+    requestObserver.onCompleted();
+    semaphore.release();
+    return new XceiverClientReply(future);
+  }
+
   private synchronized void checkOpen(DatanodeDetails dn)
       throws IOException {
     if (closed) {

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java

@@ -124,6 +124,7 @@ public BlockInputStream(
       OzoneClientConfig config) throws IOException {
     this.blockInfo = blockInfo;
     this.blockID = blockInfo.getBlockID();
+    LOG.debug("Initializing BlockInputStream for block {}", blockID);
     this.length = blockInfo.getLength();
     setPipeline(pipeline);
     tokenRef.set(token);