Merge remote-tracking branch 'upstream/master' into HDDS-13919-conditional-writes

Author: peterxcli

.github/workflows/intermittent-test-check.yml

@@ -129,7 +129,7 @@ jobs:
           java-version: ${{ github.event.inputs.java-version }}
       - name: Build (most) of Ozone
         run: |
-          args="-DskipRecon -DskipShade -Dmaven.javadoc.skip=true"
+          args="-DskipRecon -DskipShade -Dmaven.javadoc.skip=true -Drocks_tools_native"
           if [[ "${{ github.event.inputs.ratis-ref }}" != "" ]]; then
             args="$args -Dratis.version=${{ needs.ratis.outputs.ratis-version }}"
             args="$args -Dratis.thirdparty.version=${{ needs.ratis.outputs.thirdparty-version }}"
@@ -199,7 +199,7 @@ jobs:
             export OZONE_REPO_CACHED=true
           fi
 
-          args="-DexcludedGroups=native|slow|unhealthy -DskipShade"
+          args="-DexcludedGroups=native|slow|unhealthy -DskipShade -Drocks_tools_native"
           if [[ "${{ github.event.inputs.ratis-ref }}" != "" ]]; then
             args="$args -Dratis.version=${{ needs.ratis.outputs.ratis-version }}"
             args="$args -Dratis.thirdparty.version=${{ needs.ratis.outputs.thirdparty-version }}"

dev-support/byteman/hadooprpc.btm

@@ -37,7 +37,7 @@ ENDRULE
 
 
 RULE Hadoop RPC source IP
-CLASS org.apache.hadoop.ipc.Server$RpcCall
+CLASS org.apache.hadoop.ipc_.Server$RpcCall
 METHOD run
 IF true
 DO link(Thread.currentThread(), "source", $0.connection.toString())

dev-support/pmd/pmd-ruleset.xml

@@ -52,4 +52,5 @@
   </rule>
 
   <exclude-pattern>.*/generated-sources/.*</exclude-pattern>
+  <exclude-pattern>.*/org/apache/hadoop/.*_/.*</exclude-pattern>
 </ruleset>

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

@@ -113,6 +113,13 @@ public class OzoneClientConfig {
       tags = ConfigTag.CLIENT)
   private long streamBufferMaxSize = 32 * 1024 * 1024;
 
+  @Config(key = "ozone.client.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 = false;
+
   @Config(key = "ozone.client.max.retries",
       defaultValue = "5",
       description = "Maximum number of retries by Ozone Client on "
@@ -151,7 +158,7 @@ public class OzoneClientConfig {
       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 = "ozone.client.bytes.per.checksum",
@@ -160,7 +167,7 @@ public class OzoneClientConfig {
       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 = "ozone.client.verify.checksum",
@@ -538,6 +545,14 @@ public int getMaxConcurrentWritePerKey() {
     return this.maxConcurrentWritePerKey;
   }
 
+  public boolean isStreamReadBlock() {
+    return streamReadBlock;
+  }
+
+  public void setStreamReadBlock(boolean streamReadBlock) {
+    this.streamReadBlock = streamReadBlock;
+  }
+
   /**
    * Enum for indicating what mode to use when combining chunk and block
    * checksums to define an aggregate FileChecksum. This should be considered

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

@@ -17,8 +17,8 @@
 
 package org.apache.hadoop.hdds.scm;
 
-import com.google.common.base.Preconditions;
 import java.io.IOException;
+import java.util.Objects;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.scm.client.ClientTrustManager;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -49,7 +49,7 @@ public XceiverClientCreator(ConfigurationSource conf, ClientTrustManager trustMa
         OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_DEFAULT);
     this.trustManager = trustManager;
     if (securityEnabled) {
-      Preconditions.checkNotNull(trustManager);
+      Objects.requireNonNull(trustManager, "trustManager == null");
     }
   }
 

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

@@ -20,7 +20,6 @@
 import static org.apache.hadoop.hdds.HddsUtils.processForDebug;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.ArrayList;
@@ -37,6 +36,7 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeID;
@@ -59,10 +59,12 @@
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.util.Time;
+import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
 import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
 import org.apache.ratis.thirdparty.io.grpc.Status;
 import org.apache.ratis.thirdparty.io.grpc.netty.GrpcSslContexts;
 import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder;
+import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
 import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContextBuilder;
 import org.slf4j.Logger;
@@ -111,8 +113,8 @@ public class XceiverClientGrpc extends XceiverClientSpi {
   public XceiverClientGrpc(Pipeline pipeline, ConfigurationSource config,
       ClientTrustManager trustManager) {
     super();
-    Preconditions.checkNotNull(pipeline);
-    Preconditions.checkNotNull(config);
+    Objects.requireNonNull(pipeline, "pipeline == null");
+    Objects.requireNonNull(config, "config == null");
     setTimeout(config.getTimeDuration(OzoneConfigKeys.
         OZONE_CLIENT_READ_TIMEOUT, OzoneConfigKeys
         .OZONE_CLIENT_READ_TIMEOUT_DEFAULT, TimeUnit.SECONDS));
@@ -384,28 +386,15 @@ private XceiverClientReply sendCommandWithTraceIDAndRetry(
         });
   }
 
-  private XceiverClientReply sendCommandWithRetry(
-      ContainerCommandRequestProto request, List<Validator> validators)
-      throws IOException {
-    ContainerCommandResponseProto responseProto = null;
-    IOException ioException = null;
+  private List<DatanodeDetails> sortDatanodes(ContainerCommandRequestProto request) throws IOException {
+    return sortDatanodes(getRequestBlockID(request), request.getCmdType());
+  }
 
-    // In case of an exception or an error, we will try to read from the
-    // datanodes in the pipeline in a round-robin fashion.
-    XceiverClientReply reply = new XceiverClientReply(null);
+  List<DatanodeDetails> sortDatanodes(DatanodeBlockID blockID, ContainerProtos.Type cmdType) throws IOException {
     List<DatanodeDetails> datanodeList = null;
 
-    DatanodeBlockID blockID = null;
-    if (request.getCmdType() == ContainerProtos.Type.GetBlock) {
-      blockID = request.getGetBlock().getBlockID();
-    } else if  (request.getCmdType() == ContainerProtos.Type.ReadChunk) {
-      blockID = request.getReadChunk().getBlockID();
-    } else if (request.getCmdType() == ContainerProtos.Type.GetSmallFile) {
-      blockID = request.getGetSmallFile().getBlock().getBlockID();
-    }
-
     if (blockID != null) {
-      if (request.getCmdType() != ContainerProtos.Type.ReadChunk) {
+      if (cmdType != ContainerProtos.Type.ReadChunk) {
         datanodeList = pipeline.getNodes();
         int getBlockDNLeaderIndex = datanodeList.indexOf(pipeline.getLeaderNode());
         if (getBlockDNLeaderIndex > 0) {
@@ -440,6 +429,33 @@ private XceiverClientReply sendCommandWithRetry(
     if (!allInService) {
       datanodeList = sortDatanodeByOperationalState(datanodeList);
     }
+    return datanodeList;
+  }
+
+  private static DatanodeBlockID getRequestBlockID(ContainerCommandRequestProto request) {
+    DatanodeBlockID blockID = null;
+    if (request.getCmdType() == ContainerProtos.Type.GetBlock) {
+      blockID = request.getGetBlock().getBlockID();
+    } else if  (request.getCmdType() == ContainerProtos.Type.ReadChunk) {
+      blockID = request.getReadChunk().getBlockID();
+    } else if (request.getCmdType() == ContainerProtos.Type.GetSmallFile) {
+      blockID = request.getGetSmallFile().getBlock().getBlockID();
+    } else if (request.getCmdType() == ContainerProtos.Type.ReadBlock) {
+      blockID = request.getReadBlock().getBlockID();
+    }
+    return blockID;
+  }
+
+  private XceiverClientReply sendCommandWithRetry(
+      ContainerCommandRequestProto request, List<Validator> validators)
+      throws IOException {
+    ContainerCommandResponseProto responseProto = null;
+    IOException ioException = null;
+
+    // In case of an exception or an error, we will try to read from the
+    // datanodes in the pipeline in a round-robin fashion.
+    XceiverClientReply reply = new XceiverClientReply(null);
+    List<DatanodeDetails> datanodeList = sortDatanodes(request);
 
     for (DatanodeDetails dn : datanodeList) {
       try {
@@ -491,11 +507,11 @@ private XceiverClientReply sendCommandWithRetry(
       reply.setResponse(CompletableFuture.completedFuture(responseProto));
       return reply;
     } else {
-      Objects.requireNonNull(ioException);
+      Objects.requireNonNull(ioException, "ioException == null");
       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);
@@ -504,6 +520,61 @@ private XceiverClientReply sendCommandWithRetry(
     }
   }
 
+  @Override
+  public void streamRead(ContainerCommandRequestProto request,
+      StreamingReadResponse streamObserver) {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("->{}, send onNext request {}",
+          streamObserver, TextFormat.shortDebugString(request.getReadBlock()));
+    }
+    streamObserver.getRequestObserver().onNext(request);
+  }
+
+  @Override
+  public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver) throws IOException {
+    final List<DatanodeDetails> datanodeList = sortDatanodes(null, ContainerProtos.Type.ReadBlock);
+    IOException lastException = null;
+    for (DatanodeDetails dn : datanodeList) {
+      try {
+        checkOpen(dn);
+        semaphore.acquire();
+        XceiverClientProtocolServiceStub stub = asyncStubs.get(dn.getID());
+        if (stub == null) {
+          throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
+        }
+        LOG.debug("initStreamRead {} on datanode {}", blockID.getContainerBlockID(), dn);
+        StreamObserver<ContainerCommandRequestProto> requestObserver = stub
+            .withDeadlineAfter(timeout, TimeUnit.SECONDS)
+            .send(streamObserver);
+        streamObserver.setStreamingReadResponse(new StreamingReadResponse(dn,
+            (ClientCallStreamObserver<ContainerCommandRequestProto>) requestObserver));
+        return;
+      } catch (IOException e) {
+        LOG.error("Failed to start streaming read to DataNode {}", dn, e);
+        semaphore.release();
+        lastException = e;
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        throw new IOException("Interrupted initStreamRead to " + dn + " for " + blockID, e);
+      }
+    }
+    if (lastException != null) {
+      throw lastException;
+    } else {
+      throw new IOException("Failed to start streaming read to any available DataNodes");
+    }
+  }
+
+  /**
+   * This method should be called to indicate the end of streaming read. Its primary purpose is to release the
+   * semaphore acquired when starting the streaming read, but is also used to update any metrics or debug logs as
+   * needed.
+   */
+  @Override
+  public void completeStreamRead() {
+    semaphore.release();
+  }
+
   private static List<DatanodeDetails> sortDatanodeByOperationalState(
       List<DatanodeDetails> datanodeList) {
     List<DatanodeDetails> sortedDatanodeList = new ArrayList<>(datanodeList);

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

@@ -29,6 +29,7 @@
 import com.google.common.cache.RemovalListener;
 import com.google.common.cache.RemovalNotification;
 import java.io.IOException;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.hdds.conf.Config;
 import org.apache.hadoop.hdds.conf.ConfigGroup;
@@ -79,8 +80,8 @@ public XceiverClientManager(ConfigurationSource conf,
       ScmClientConfig clientConf,
       ClientTrustManager trustManager) throws IOException {
     super(conf, trustManager);
-    Preconditions.checkNotNull(clientConf);
-    Preconditions.checkNotNull(conf);
+    Objects.requireNonNull(clientConf, "clientConf == null");
+    Objects.requireNonNull(conf, "conf == null");
     long staleThresholdMs = clientConf.getStaleThreshold(MILLISECONDS);
 
     this.clientCache = CacheBuilder.newBuilder()
@@ -118,7 +119,7 @@ public Cache<String, XceiverClientSpi> getClientCache() {
   @Override
   public XceiverClientSpi acquireClient(Pipeline pipeline,
       boolean topologyAware) throws IOException {
-    Preconditions.checkNotNull(pipeline);
+    Objects.requireNonNull(pipeline, "pipeline == null");
     Preconditions.checkArgument(pipeline.getNodes() != null);
     Preconditions.checkArgument(!pipeline.getNodes().isEmpty(),
         NO_REPLICA_FOUND);
@@ -133,7 +134,7 @@ public XceiverClientSpi acquireClient(Pipeline pipeline,
   @Override
   public void releaseClient(XceiverClientSpi client, boolean invalidateClient,
       boolean topologyAware) {
-    Preconditions.checkNotNull(client);
+    Objects.requireNonNull(client, "client == null");
     synchronized (clientCache) {
       client.decrementReference();
       if (invalidateClient) {
@@ -246,7 +247,7 @@ public static synchronized void resetXceiverClientMetrics() {
   @ConfigGroup(prefix = "scm.container.client")
   public static class ScmClientConfig {
 
-    @Config(key = "max.size",
+    @Config(key = "scm.container.client.max.size",
         defaultValue = "256",
         tags = {OZONE, PERFORMANCE},
         description =
@@ -257,7 +258,7 @@ public static class ScmClientConfig {
     )
     private int maxSize;
 
-    @Config(key = "idle.threshold",
+    @Config(key = "scm.container.client.idle.threshold",
         type = ConfigType.TIME, timeUnit = MILLISECONDS,
         defaultValue = "10s",
         tags = {OZONE, PERFORMANCE},

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

@@ -392,8 +392,7 @@ public XceiverClientReply sendCommandAsync(
               // able to connect to leader in the pipeline, though the
               // pipeline can still be functional.
               RaftException exception = reply.getException();
-              Preconditions.checkNotNull(exception, "Raft reply failure but " +
-                  "no exception propagated.");
+              Objects.requireNonNull(exception, "Raft reply failure but no exception propagated.");
               throw new CompletionException(exception);
             }
             ContainerCommandResponseProto response =

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

@@ -17,11 +17,11 @@
 
 package org.apache.hadoop.hdds.scm.client;
 
-import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
@@ -219,7 +219,7 @@ public static void verifyKeyName(String keyName) {
    */
   public static <T> void checkNotNull(T... references) {
     for (T ref: references) {
-      Preconditions.checkNotNull(ref);
+      Objects.requireNonNull(ref, "ref == null");
     }
   }
 

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

@@ -105,7 +105,7 @@ protected void handleRetry(IOException exception, RetryPolicy retryPolicy)
     if (Thread.currentThread().isInterrupted()) {
       setExceptionAndThrow(exception);
     }
-    Objects.requireNonNull(action);
+    Objects.requireNonNull(action, "action == null");
     Preconditions.checkArgument(
         action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
     if (action.delayMillis > 0) {