rebase

Author: hanishakoneru

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

@@ -19,47 +19,27 @@
 package org.apache.hadoop.hdds.scm.storage;
 
 import com.google.common.annotations.VisibleForTesting;
-<<<<<<< HEAD
-import com.google.common.base.Preconditions;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.scm.container.common.helpers
-    .StorageContainerException;
-import org.apache.hadoop.ozone.common.Checksum;
-import org.apache.hadoop.ozone.common.ChecksumData;
-import org.apache.hadoop.ozone.common.OzoneChecksumException;
-import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
-=======
+
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
->>>>>>> Partial Chunk reads
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
-<<<<<<< HEAD
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
-    .ReadChunkResponseProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.
-    ContainerCommandResponseProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.
-    ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.client.BlockID;
-=======
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.GetBlockResponseProto;
-import org.apache.hadoop.hdds.client.BlockID;
-import org.apache.ratis.util.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
->>>>>>> Partial Chunk reads
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -103,18 +83,6 @@ public class BlockInputStream extends InputStream implements Seekable {
   // Index of the chunkStream corresponding to the current position of the
   // BlockInputStream i.e offset of the data to be read next from this block
   private int chunkIndex;
-<<<<<<< HEAD
-  // ChunkIndexOfCurrentBuffer points to the index of chunk read into the
-  // buffers or index of the last chunk in the buffers. It is updated only
-  // when a new chunk is read from container into the buffers.
-  private int chunkIndexOfCurrentBuffer;
-  private long[] chunkOffset;
-  private List<ByteBuffer> buffers;
-  private int bufferIndex;
-  private long bufferPosition;
-  private boolean verifyChecksum;
-=======
->>>>>>> Partial Chunk reads
 
   // Position of the BlockInputStream is maintainted by this variable till
   // the stream is initialized. This position is w.r.t to the block only and
@@ -241,37 +209,13 @@ public synchronized int read(byte[] b, int off, int len) throws IOException {
     if (off < 0 || len < 0 || len > b.length - off) {
       throw new IndexOutOfBoundsException();
     }
-<<<<<<< HEAD
-    // ChunkIndex is the last chunk in the stream. Check if this chunk has
-    // been read from container or not. Return true if chunkIndex has not
-    // been read yet and false otherwise.
-    return chunkIndexOfCurrentBuffer != chunkIndex;
-  }
-
-  /**
-   * Attempts to read the chunk at the specified offset in the chunk list.  If
-   * successful, then the data of the read chunk is saved so that its bytes can
-   * be returned from subsequent read calls.
-   *
-   * @throws IOException if there is an I/O error while performing the call
-   */
-  private synchronized void readChunkFromContainer() throws IOException {
-    // Read the chunk at chunkIndex
-    final ChunkInfo chunkInfo = chunks.get(chunkIndex);
-    ByteString byteString;
-    byteString = readChunk(chunkInfo);
-    buffers = byteString.asReadOnlyByteBufferList();
-    bufferIndex = 0;
-    chunkIndexOfCurrentBuffer = chunkIndex;
-=======
     if (len == 0) {
       return 0;
     }
 
     if (!initialized) {
       initialize();
     }
->>>>>>> Partial Chunk reads
 
     checkOpen();
     int totalReadLen = 0;
@@ -284,26 +228,6 @@ private synchronized void readChunkFromContainer() throws IOException {
         return totalReadLen == 0 ? EOF : totalReadLen;
       }
 
-<<<<<<< HEAD
-  /**
-   * Send RPC call to get the chunk from the container.
-   */
-  @VisibleForTesting
-  protected ByteString readChunk(final ChunkInfo chunkInfo)
-      throws IOException {
-    ReadChunkResponseProto readChunkResponse;
-    try {
-      List<CheckedBiFunction> validators =
-          ContainerProtocolCalls.getValidatorList();
-      validators.add(validator);
-      readChunkResponse = ContainerProtocolCalls
-          .readChunk(xceiverClient, chunkInfo, blockID, traceID, validators);
-    } catch (IOException e) {
-      if (e instanceof StorageContainerException) {
-        throw e;
-      }
-      throw new IOException("Unexpected OzoneException: " + e.toString(), e);
-=======
       // Get the current chunkStream and read data from it
       ChunkInputStream current = chunkStreams.get(chunkIndex);
       int numBytesToRead = Math.min(len, (int)current.getRemaining());
@@ -323,33 +247,10 @@ protected ByteString readChunk(final ChunkInfo chunkInfo)
           ((chunkIndex + 1) < chunkStreams.size())) {
         chunkIndex += 1;
       }
->>>>>>> Partial Chunk reads
     }
     return totalReadLen;
   }
 
-<<<<<<< HEAD
-  private CheckedBiFunction<ContainerCommandRequestProto,
-      ContainerCommandResponseProto, IOException> validator =
-          (request, response) -> {
-            ReadChunkResponseProto readChunkResponse = response.getReadChunk();
-            final ChunkInfo chunkInfo = readChunkResponse.getChunkData();
-            ByteString byteString = readChunkResponse.getData();
-            if (byteString.size() != chunkInfo.getLen()) {
-              // Bytes read from chunk should be equal to chunk size.
-              throw new OzoneChecksumException(String
-                  .format("Inconsistent read for chunk=%s len=%d bytesRead=%d",
-                      chunkInfo.getChunkName(), chunkInfo.getLen(),
-                      byteString.size()));
-            }
-            ChecksumData checksumData =
-                ChecksumData.getFromProtoBuf(chunkInfo.getChecksumData());
-            if (verifyChecksum) {
-              Checksum.verifyChecksum(byteString, checksumData);
-            }
-          };
-
-=======
   /**
    * Seeks the BlockInputStream to the specified position. If the stream is
    * not initialized, save the seeked position via blockPosition. Otherwise,
@@ -366,7 +267,6 @@ protected ByteString readChunk(final ChunkInfo chunkInfo)
    *    2. chunkStream[2] will be seeked to position 10
    *       (= 90 - chunkOffset[2] (= 80)).
    */
->>>>>>> Partial Chunk reads
   @Override
   public synchronized void seek(long pos) throws IOException {
     if (!initialized) {
@@ -386,7 +286,7 @@ public synchronized void seek(long pos) throws IOException {
       throw new EOFException(
           "EOF encountered at pos: " + pos + " for block: " + blockID);
     }
-    Preconditions.assertTrue(chunkIndex >= 0);
+
     if (chunkIndex >= chunkStreams.size()) {
       chunkIndex = Arrays.binarySearch(chunkOffsets, pos);
     } else if (pos < chunkOffsets[chunkIndex]) {

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

@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+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.ChunkInfo;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto;
@@ -31,15 +32,14 @@
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.common.Checksum;
 import org.apache.hadoop.ozone.common.ChecksumData;
+import org.apache.hadoop.ozone.common.OzoneChecksumException;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 
 /**
  * An {@link InputStream} used by the REST service in combination with the
@@ -56,7 +56,7 @@ public class ChunkInputStream extends InputStream implements Seekable {
   private final BlockID blockID;
   private final String traceID;
   private XceiverClientSpi xceiverClient;
-  private final boolean verifyChecksum;
+  private boolean verifyChecksum;
   private boolean allocated = false;
 
   // Buffer to store the chunk data read from the DN container
@@ -275,10 +275,6 @@ private synchronized int prepareRead(int len) throws IOException {
    */
   private synchronized void readChunkFromContainer(int len) throws IOException {
 
-    List<DatanodeDetails> excludeDns = null;
-    ByteString byteString;
-    List<DatanodeDetails> dnList = getDatanodeList();
-
     // index of first byte to be read from the chunk
     long startByteIndex;
     if (chunkPosition >= 0) {
@@ -307,41 +303,7 @@ private synchronized void readChunkFromContainer(int len) throws IOException {
         .setLen(bufferLength)
         .build();
 
-    while (true) {
-      List<DatanodeDetails> dnListFromReadChunkCall = new ArrayList<>();
-      byteString = readChunk(adjustedChunkInfo, excludeDns,
-          dnListFromReadChunkCall);
-      try {
-        if (byteString.size() != adjustedChunkInfo.getLen()) {
-          // Bytes read from chunk should be equal to chunk size.
-          throw new IOException(String
-              .format("Inconsistent read for chunk=%s len=%d bytesRead=%d",
-                  adjustedChunkInfo.getChunkName(), adjustedChunkInfo.getLen(),
-                  byteString.size()));
-        }
-
-        if (verifyChecksum) {
-          ChecksumData checksumData = ChecksumData.getFromProtoBuf(
-              chunkInfo.getChecksumData());
-          int checkumStartIndex =
-              (int) (bufferOffset / checksumData.getBytesPerChecksum());
-          Checksum.verifyChecksum(byteString, checksumData, checkumStartIndex);
-        }
-        break;
-      } catch (IOException ioe) {
-        // we will end up in this situation only if the checksum mismatch
-        // happens or the length of the chunk mismatches.
-        // In this case, read should be retried on a different replica.
-        // TODO: Inform SCM of a possible corrupt container replica here
-        if (excludeDns == null) {
-          excludeDns = new ArrayList<>();
-        }
-        excludeDns.addAll(dnListFromReadChunkCall);
-        if (excludeDns.size() == dnList.size()) {
-          throw ioe;
-        }
-      }
-    }
+    ByteString byteString = readChunk(adjustedChunkInfo);
 
     buffers = byteString.asReadOnlyByteBufferList();
     bufferIndex = 0;
@@ -361,37 +323,51 @@ private synchronized void readChunkFromContainer(int len) throws IOException {
    * Send RPC call to get the chunk from the container.
    */
   @VisibleForTesting
-  protected ByteString readChunk(ChunkInfo readChunkInfo,
-      List<DatanodeDetails> excludeDns,
-      List<DatanodeDetails> dnListFromReply) throws IOException {
-    XceiverClientReply reply;
+  protected ByteString readChunk(ChunkInfo readChunkInfo) throws IOException {
     ReadChunkResponseProto readChunkResponse;
 
     try {
-      reply = ContainerProtocolCalls.readChunk(xceiverClient, readChunkInfo,
-          blockID, traceID, excludeDns);
+      List<CheckedBiFunction> validators =
+          ContainerProtocolCalls.getValidatorList();
+      validators.add(validator);
 
-      ContainerCommandResponseProto response = reply.getResponse().get();
-      ContainerProtocolCalls.validateContainerResponse(response);
-      readChunkResponse = response.getReadChunk();
+      readChunkResponse = ContainerProtocolCalls.readChunk(xceiverClient,
+          readChunkInfo, blockID, traceID, validators);
 
-      dnListFromReply.addAll(reply.getDatanodes());
     } catch (IOException e) {
       if (e instanceof StorageContainerException) {
         throw e;
       }
       throw new IOException("Unexpected OzoneException: " + e.toString(), e);
-    } catch (ExecutionException | InterruptedException e) {
-      throw new IOException("Failed to execute ReadChunk command for chunk "
-          + readChunkInfo.getChunkName(), e);
     }
+
     return readChunkResponse.getData();
   }
 
-  @VisibleForTesting
-  protected List<DatanodeDetails> getDatanodeList() {
-    return xceiverClient.getPipeline().getNodes();
-  }
+  private CheckedBiFunction<ContainerCommandRequestProto,
+      ContainerCommandResponseProto, IOException> validator =
+      (request, response) -> {
+        final ChunkInfo chunkInfo = request.getReadChunk().getChunkData();
+
+        ReadChunkResponseProto readChunkResponse = response.getReadChunk();
+        ByteString byteString = readChunkResponse.getData();
+
+        if (byteString.size() != chunkInfo.getLen()) {
+          // Bytes read from chunk should be equal to chunk size.
+          throw new OzoneChecksumException(String
+              .format("Inconsistent read for chunk=%s len=%d bytesRead=%d",
+                  chunkInfo.getChunkName(), chunkInfo.getLen(),
+                  byteString.size()));
+        }
+
+        if (verifyChecksum) {
+          ChecksumData checksumData =
+              ChecksumData.getFromProtoBuf(chunkInfo.getChecksumData());
+          int checkumStartIndex =
+              (int) (bufferOffset / checksumData.getBytesPerChecksum());
+          Checksum.verifyChecksum(byteString, checksumData, checkumStartIndex);
+        }
+      };
 
   /**
    * Return the offset and length of bytes that need to be read from the

hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java

@@ -120,14 +120,8 @@ private class DummyBlockInputStream extends BlockInputStream {
     }
 
     @Override
-<<<<<<< HEAD
-    protected ByteString readChunk(final ChunkInfo chunkInfo)
-        throws IOException {
-      return getByteString(chunkInfo.getChunkName(), (int) chunkInfo.getLen());
-=======
     protected List<ChunkInfo> getChunkInfos() {
       return chunks;
->>>>>>> Partial Chunk reads
     }
 
     @Override

hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java

@@ -103,22 +103,14 @@ public DummyChunkInputStream(ChunkInfo chunkInfo,
     }
 
     @Override
-    protected ByteString readChunk(ChunkInfo readChunkInfo,
-        List<DatanodeDetails> excludeDns,
-        List<DatanodeDetails> dnListFromReply) {
+    protected ByteString readChunk(ChunkInfo readChunkInfo) {
       ByteString byteString = ByteString.copyFrom(chunkData,
           (int) readChunkInfo.getOffset(),
           (int) readChunkInfo.getLen());
       readByteBuffers.add(byteString);
       return byteString;
     }
 
-    @Override
-    protected List<DatanodeDetails> getDatanodeList() {
-      // return an empty dummy list of size 5
-      return new ArrayList<>(5);
-    }
-
     @Override
     protected void checkOpen() {
       // No action needed