[DO NOT MERGE] Backport some fixes, performance optimizations from master to ozone-1.4 #6929

.github/workflows/ci.yml

@@ -135,13 +135,17 @@ jobs:
       - build-info
       - build
       - basic
-    runs-on: ubuntu-20.04
     timeout-minutes: 30
     if: needs.build-info.outputs.needs-compile == 'true'
     strategy:
       matrix:
         java: [ 11, 17 ]
+        include:
+          - os: ubuntu-20.04
+          - java: 8
+            os: macos-12
       fail-fast: false
+    runs-on: ${{ matrix.os }}
     steps:
       - name: Download Ozone source tarball
         uses: actions/download-artifact@v4

dev-support/ci/selective_ci_checks.bats

@@ -177,17 +177,20 @@ load bats-assert/load.bash
   assert_output -p needs-kubernetes-tests=false
 }
 
-@test "native test in other module" {
-  run dev-support/ci/selective_ci_checks.sh 7d01cc14a6
-
-  assert_output -p 'basic-checks=["rat","author","checkstyle","findbugs","native","unit"]'
-  assert_output -p needs-build=true
-  assert_output -p needs-compile=true
-  assert_output -p needs-compose-tests=false
-  assert_output -p needs-dependency-check=false
-  assert_output -p needs-integration-tests=false
-  assert_output -p needs-kubernetes-tests=false
-}
+# disabled, because this test fails if
+# hadoop-hdds/rocksdb-checkpoint-differ/src/test/java/org/apache/ozone/rocksdb/util/TestManagedSstFileReader.java
+# is not present in the current tree (i.e. if file is renamed, moved or deleted)
+#@test "native test in other module" {
+#  run dev-support/ci/selective_ci_checks.sh 7d01cc14a6
+#
+#  assert_output -p 'basic-checks=["rat","author","checkstyle","findbugs","native","unit"]'
+#  assert_output -p needs-build=true
+#  assert_output -p needs-compile=true
+#  assert_output -p needs-compose-tests=false
+#  assert_output -p needs-dependency-check=false
+#  assert_output -p needs-integration-tests=false
+#  assert_output -p needs-kubernetes-tests=false
+#}
 
 @test "kubernetes only" {
   run dev-support/ci/selective_ci_checks.sh 5336bb9bd

hadoop-hdds/client/pom.xml

@@ -55,7 +55,7 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
     </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
-      <artifactId>mockito-core</artifactId>
+      <artifactId>mockito-inline</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>

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

@@ -144,6 +144,23 @@ public enum ChecksumCombineMode {
       tags = ConfigTag.CLIENT)
   private int retryInterval = 0;
 
+  @Config(key = "read.max.retries",
+      defaultValue = "3",
+      description = "Maximum number of retries by Ozone Client on "
+          + "encountering connectivity exception when reading a key.",
+      tags = ConfigTag.CLIENT)
+  private int maxReadRetryCount = 3;
+
+  @Config(key = "read.retry.interval",
+      defaultValue = "1",
+      description =
+          "Indicates the time duration in seconds a client will wait "
+              + "before retrying a read key request on encountering "
+              + "a connectivity excepetion from Datanodes . "
+              + "By default the interval is 1 second",
+      tags = ConfigTag.CLIENT)
+  private int readRetryInterval = 1;
+
   @Config(key = "checksum.type",
       defaultValue = "CRC32",
       description = "The checksum type [NONE/ CRC32/ CRC32C/ SHA256/ MD5] "
@@ -326,6 +343,22 @@ public void setRetryInterval(int retryInterval) {
     this.retryInterval = retryInterval;
   }
 
+  public int getMaxReadRetryCount() {
+    return maxReadRetryCount;
+  }
+
+  public void setMaxReadRetryCount(int maxReadRetryCount) {
+    this.maxReadRetryCount = maxReadRetryCount;
+  }
+
+  public int getReadRetryInterval() {
+    return readRetryInterval;
+  }
+
+  public void setReadRetryInterval(int readRetryInterval) {
+    this.readRetryInterval = readRetryInterval;
+  }
+
   public ChecksumType getChecksumType() {
     return ChecksumType.valueOf(checksumType);
   }

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

@@ -49,6 +49,7 @@
 import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
 import org.apache.hadoop.hdds.tracing.GrpcClientInterceptor;
 import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
 import java.util.concurrent.TimeoutException;
@@ -274,6 +275,11 @@ public ContainerCommandResponseProto sendCommand(
     List<DatanodeDetails> datanodeList = pipeline.getNodes();
     HashMap<DatanodeDetails, CompletableFuture<ContainerCommandResponseProto>>
             futureHashMap = new HashMap<>();
+    if (!request.hasVersion()) {
+      ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto.newBuilder(request);
+      builder.setVersion(ClientVersion.CURRENT.toProtoValue());
+      request = builder.build();
+    }
     for (DatanodeDetails dn : datanodeList) {
       try {
         futureHashMap.put(dn, sendCommandAsync(request, dn).getResponse());
@@ -334,10 +340,13 @@ private XceiverClientReply sendCommandWithTraceIDAndRetry(
 
     return TracingUtil.executeInNewSpan(spanName,
         () -> {
-          ContainerCommandRequestProto finalPayload =
+          ContainerCommandRequestProto.Builder builder =
               ContainerCommandRequestProto.newBuilder(request)
-                  .setTraceID(TracingUtil.exportCurrentSpan()).build();
-          return sendCommandWithRetry(finalPayload, validators);
+                  .setTraceID(TracingUtil.exportCurrentSpan());
+          if (!request.hasVersion()) {
+            builder.setVersion(ClientVersion.CURRENT.toProtoValue());
+          }
+          return sendCommandWithRetry(builder.build(), validators);
         });
   }
 
@@ -457,12 +466,14 @@ public XceiverClientReply sendCommandAsync(
 
     try (Scope ignored = GlobalTracer.get().activateSpan(span)) {
 
-      ContainerCommandRequestProto finalPayload =
+      ContainerCommandRequestProto.Builder builder =
           ContainerCommandRequestProto.newBuilder(request)
-              .setTraceID(TracingUtil.exportCurrentSpan())
-              .build();
+              .setTraceID(TracingUtil.exportCurrentSpan());
+      if (!request.hasVersion()) {
+        builder.setVersion(ClientVersion.CURRENT.toProtoValue());
+      }
       XceiverClientReply asyncReply =
-          sendCommandAsync(finalPayload, pipeline.getFirstNode());
+          sendCommandAsync(builder.build(), pipeline.getFirstNode());
       if (shouldBlockAndWaitAsyncReply(request)) {
         asyncReply.getResponse().get();
       }

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

@@ -36,6 +36,7 @@
 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.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi.Validator;
@@ -76,8 +77,8 @@ public class BlockInputStream extends BlockExtendedInputStream {
   private XceiverClientSpi xceiverClient;
   private boolean initialized = false;
   // TODO: do we need to change retrypolicy based on exception.
-  private final RetryPolicy retryPolicy =
-      HddsClientUtils.createRetryPolicy(3, TimeUnit.SECONDS.toMillis(1));
+  private final RetryPolicy retryPolicy;
+
   private int retries;
 
   // List of ChunkInputStreams, one for each chunk in the block
@@ -112,25 +113,29 @@ public class BlockInputStream extends BlockExtendedInputStream {
   private final Function<BlockID, BlockLocationInfo> refreshFunction;
 
   public BlockInputStream(BlockID blockId, long blockLen, Pipeline pipeline,
-      Token<OzoneBlockTokenIdentifier> token, boolean verifyChecksum,
+      Token<OzoneBlockTokenIdentifier> token,
       XceiverClientFactory xceiverClientFactory,
-      Function<BlockID, BlockLocationInfo> refreshFunction) {
+      Function<BlockID, BlockLocationInfo> refreshFunction,
+      OzoneClientConfig config) throws IOException {
     this.blockID = blockId;
     this.length = blockLen;
     setPipeline(pipeline);
     tokenRef.set(token);
-    this.verifyChecksum = verifyChecksum;
+    this.verifyChecksum = config.isChecksumVerify();
     this.xceiverClientFactory = xceiverClientFactory;
     this.refreshFunction = refreshFunction;
+    this.retryPolicy =
+        HddsClientUtils.createRetryPolicy(config.getMaxReadRetryCount(),
+            TimeUnit.SECONDS.toMillis(config.getReadRetryInterval()));
   }
 
   public BlockInputStream(BlockID blockId, long blockLen, Pipeline pipeline,
                           Token<OzoneBlockTokenIdentifier> token,
-                          boolean verifyChecksum,
-                          XceiverClientFactory xceiverClientFactory
-  ) {
-    this(blockId, blockLen, pipeline, token, verifyChecksum,
-        xceiverClientFactory, null);
+                          XceiverClientFactory xceiverClientFactory,
+                          OzoneClientConfig config
+  ) throws IOException {
+    this(blockId, blockLen, pipeline, token,
+        xceiverClientFactory, null, config);
   }
   /**
    * Initialize the BlockInputStream. Get the BlockData (list of chunks) from
@@ -239,33 +244,28 @@ protected List<ChunkInfo> getChunkInfoList() throws IOException {
 
   @VisibleForTesting
   protected List<ChunkInfo> getChunkInfoListUsingClient() throws IOException {
-    final Pipeline pipeline = xceiverClient.getPipeline();
-
+    Pipeline pipeline = pipelineRef.get();
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Initializing BlockInputStream for get key to access {}",
-          blockID.getContainerID());
-    }
-
-    DatanodeBlockID.Builder blkIDBuilder =
-        DatanodeBlockID.newBuilder().setContainerID(blockID.getContainerID())
-            .setLocalID(blockID.getLocalID())
-            .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId());
-
-    int replicaIndex = pipeline.getReplicaIndex(pipeline.getClosestNode());
-    if (replicaIndex > 0) {
-      blkIDBuilder.setReplicaIndex(replicaIndex);
+      LOG.debug("Initializing BlockInputStream for get key to access {} with pipeline {}.",
+          blockID.getContainerID(), pipeline);
     }
 
     GetBlockResponseProto response = ContainerProtocolCalls.getBlock(
-        xceiverClient, VALIDATORS, blkIDBuilder.build(), tokenRef.get());
+        xceiverClient, VALIDATORS, blockID, tokenRef.get(), pipeline.getReplicaIndexes());
 
     return response.getBlockData().getChunksList();
   }
 
-  private void setPipeline(Pipeline pipeline) {
+  private void setPipeline(Pipeline pipeline) throws IOException {
     if (pipeline == null) {
       return;
     }
+    long replicaIndexes = pipeline.getNodes().stream().mapToInt(pipeline::getReplicaIndex).distinct().count();
+
+    if (replicaIndexes > 1) {
+      throw new IOException(String.format("Pipeline: %s has nodes containing different replica indexes.",
+          pipeline));
+    }
 
     // irrespective of the container state, we will always read via Standalone
     // protocol.

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

@@ -26,6 +26,8 @@
 import org.apache.hadoop.fs.CanUnbuffer;
 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;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
@@ -60,6 +62,7 @@ public class ChunkInputStream extends InputStream
   private final ChunkInfo chunkInfo;
   private final long length;
   private final BlockID blockID;
+  private ContainerProtos.DatanodeBlockID datanodeBlockID;
   private final XceiverClientFactory xceiverClientFactory;
   private XceiverClientSpi xceiverClient;
   private final Supplier<Pipeline> pipelineSupplier;
@@ -290,13 +293,27 @@ protected synchronized void releaseClient() {
     }
   }
 
+  /**
+   * Updates DatanodeBlockId which based on blockId.
+   */
+  private void updateDatanodeBlockId(Pipeline pipeline) throws IOException {
+    DatanodeDetails closestNode = pipeline.getClosestNode();
+    int replicaIdx = pipeline.getReplicaIndex(closestNode);
+    ContainerProtos.DatanodeBlockID.Builder builder = blockID.getDatanodeBlockIDProtobufBuilder();
+    if (replicaIdx > 0) {
+      builder.setReplicaIndex(replicaIdx);
+    }
+    datanodeBlockID = builder.build();
+  }
+
   /**
    * Acquire new client if previous one was released.
    */
   protected synchronized void acquireClient() throws IOException {
     if (xceiverClientFactory != null && xceiverClient == null) {
-      xceiverClient = xceiverClientFactory.acquireClientForReadData(
-          pipelineSupplier.get());
+      Pipeline pipeline = pipelineSupplier.get();
+      xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline);
+      updateDatanodeBlockId(pipeline);
     }
   }
 
@@ -422,8 +439,8 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
       throws IOException {
 
     ReadChunkResponseProto readChunkResponse =
-        ContainerProtocolCalls.readChunk(xceiverClient,
-            readChunkInfo, blockID, validators, tokenSupplier.get());
+        ContainerProtocolCalls.readChunk(xceiverClient, readChunkInfo, datanodeBlockID, validators,
+            tokenSupplier.get());
 
     if (readChunkResponse.hasData()) {
       return readChunkResponse.getData().asReadOnlyByteBufferList()

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

@@ -115,17 +115,28 @@ ContainerCommandResponseProto> executePutBlock(boolean close,
     }
 
     BlockData checksumBlockData = null;
+    BlockID blockID = null;
     //Reverse Traversal as all parity will have checksumBytes
     for (int i = blockData.length - 1; i >= 0; i--) {
       BlockData bd = blockData[i];
       if (bd == null) {
         continue;
       }
+      if (blockID == null) {
+        // store the BlockID for logging
+        blockID = bd.getBlockID();
+      }
       List<ChunkInfo> chunks = bd.getChunks();
-      if (chunks != null && chunks.size() > 0 && chunks.get(0)
-          .hasStripeChecksum()) {
-        checksumBlockData = bd;
-        break;
+      if (chunks != null && chunks.size() > 0) {
+        if (chunks.get(0).hasStripeChecksum()) {
+          checksumBlockData = bd;
+          break;
+        } else {
+          ChunkInfo chunk = chunks.get(0);
+          LOG.debug("The first chunk in block with index {} does not have stripeChecksum. BlockID: {}, Block " +
+                  "size: {}. Chunk length: {}, Chunk offset: {}, hasChecksumData: {}, chunks size: {}.", i,
+              bd.getBlockID(), bd.getSize(), chunk.getLen(), chunk.getOffset(), chunk.hasChecksumData(), chunks.size());
+        }
       }
     }
 
@@ -158,9 +169,8 @@ ContainerCommandResponseProto> executePutBlock(boolean close,
       getContainerBlockData().clearChunks();
       getContainerBlockData().addAllChunks(newChunkList);
     } else {
-      throw new IOException("None of the block data have checksum " +
-          "which means " + parity + "(parity)+1 blocks are " +
-          "not present");
+      LOG.warn("Could not find checksum data in any index for blockData with BlockID {}, length {} and " +
+          "blockGroupLength {}.", blockID, blockData.length, blockGroupLength);
     }
 
     return executePutBlock(close, force, blockGroupLength);

hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java

@@ -19,13 +19,15 @@
 
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
 import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.security.token.Token;
 
+import java.io.IOException;
 import java.util.function.Function;
 
 /**
@@ -48,8 +50,9 @@ public interface BlockInputStreamFactory {
    */
   BlockExtendedInputStream create(ReplicationConfig repConfig,
       BlockLocationInfo blockInfo, Pipeline pipeline,
-      Token<OzoneBlockTokenIdentifier> token, boolean verifyChecksum,
+      Token<OzoneBlockTokenIdentifier> token,
        XceiverClientFactory xceiverFactory,
-       Function<BlockID, BlockLocationInfo> refreshFunction);
+       Function<BlockID, BlockLocationInfo> refreshFunction,
+       OzoneClientConfig config) throws IOException;
 
 }