Merge branch 'master' into HDDS-5713

Author: szetszwo

.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/ci/selective_ci_checks.bats

@@ -309,6 +309,17 @@ load bats-assert/load.bash
   assert_output -p needs-kubernetes-tests=false
 }
 
+@test "CI workflow change (check.yml)" {
+  run dev-support/ci/selective_ci_checks.sh 1468af02067ec75b255f605816c32f8bf4dfaabf
+
+  assert_output -p 'basic-checks=["author","bats","checkstyle","docs","findbugs","pmd","rat"]'
+  assert_output -p needs-build=true
+  assert_output -p needs-compile=true
+  assert_output -p needs-compose-tests=true
+  assert_output -p needs-integration-tests=true
+  assert_output -p needs-kubernetes-tests=true
+}
+
 @test "CI workflow change (ci.yaml)" {
   run dev-support/ci/selective_ci_checks.sh 90fd5f2adc
 

dev-support/ci/selective_ci_checks.sh

@@ -192,6 +192,7 @@ function check_if_tests_are_needed_at_all() {
 function run_all_tests_if_environment_files_changed() {
     start_end::group_start "Check if everything should be run"
     local pattern_array=(
+        "^.github/workflows/check.yml"
         "^.github/workflows/ci.yml"
         "^.github/workflows/post-commit.yml"
         "^dev-support/ci"

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdds.scm;
 
 import com.google.common.base.Preconditions;
+import java.time.Duration;
 import org.apache.hadoop.hdds.conf.Config;
 import org.apache.hadoop.hdds.conf.ConfigGroup;
 import org.apache.hadoop.hdds.conf.ConfigTag;
@@ -281,6 +282,27 @@ public class OzoneClientConfig {
       tags = ConfigTag.CLIENT)
   private int maxConcurrentWritePerKey = 1;
 
+  @Config(key = "ozone.client.stream.read.pre-read-size",
+      defaultValue = "33554432",
+      type = ConfigType.LONG,
+      tags = {ConfigTag.CLIENT},
+      description = "Extra bytes to prefetch during streaming reads.")
+  private long streamReadPreReadSize = 32L << 20;
+
+  @Config(key = "ozone.client.stream.read.response-data-size",
+      defaultValue = "1048576",
+      type = ConfigType.INT,
+      tags = {ConfigTag.CLIENT},
+      description = "Chunk size of streaming read responses from datanodes.")
+  private int streamReadResponseDataSize = 1 << 20;
+
+  @Config(key = "ozone.client.stream.read.timeout",
+      defaultValue = "10s",
+      type = ConfigType.TIME,
+      tags = {ConfigTag.CLIENT},
+      description = "Timeout for receiving streaming read responses.")
+  private Duration streamReadTimeout = Duration.ofSeconds(10);
+
   @PostConstruct
   public void validate() {
     Preconditions.checkState(streamBufferSize > 0);
@@ -335,6 +357,33 @@ public void validate() {
       }
       // Note: ozone.fs.hsync.enabled is enforced by OzoneFSUtils#canEnableHsync, not here
     }
+    // Validate streaming read configurations.
+    // Ensure pre-read size is non-negative. If it's invalid, reset to a sane default.
+    if (streamReadPreReadSize < 0) {
+      LOG.warn("Invalid ozone.client.stream.read.pre-read-size = {}. " +
+              "Resetting to default 32MB.",
+          streamReadPreReadSize);
+      streamReadPreReadSize = 32L << 20; // 32MB
+    }
+
+    // Ensure response data size is positive.
+    if (streamReadResponseDataSize <= 0) {
+      LOG.warn("Invalid ozone.client.stream.read.response-data-size = {}. " +
+              "Resetting to default 1MB.",
+          streamReadResponseDataSize);
+      streamReadResponseDataSize = 1 << 20; // 1MB
+    }
+
+    // Ensure stream read timeout is a positive duration.
+    Duration defaultTimeout = Duration.ofSeconds(10);
+    if (streamReadTimeout == null
+        || streamReadTimeout.isZero()
+        || streamReadTimeout.isNegative()) {
+      LOG.warn("Invalid ozone.client.stream.read.timeout = {}. " +
+              "Resetting to default {}.",
+          streamReadTimeout, defaultTimeout);
+      streamReadTimeout = defaultTimeout;
+    }
   }
 
   public long getStreamBufferFlushSize() {
@@ -553,6 +602,30 @@ public void setStreamReadBlock(boolean streamReadBlock) {
     this.streamReadBlock = streamReadBlock;
   }
 
+  public long getStreamReadPreReadSize() {
+    return streamReadPreReadSize;
+  }
+
+  public int getStreamReadResponseDataSize() {
+    return streamReadResponseDataSize;
+  }
+
+  public Duration getStreamReadTimeout() {
+    return streamReadTimeout;
+  }
+
+  public void setStreamReadPreReadSize(long streamReadPreReadSize) {
+    this.streamReadPreReadSize = streamReadPreReadSize;
+  }
+
+  public void setStreamReadResponseDataSize(int streamReadResponseDataSize) {
+    this.streamReadResponseDataSize = streamReadResponseDataSize;
+  }
+
+  public void setStreamReadTimeout(Duration streamReadTimeout) {
+    this.streamReadTimeout = streamReadTimeout;
+  }
+
   /**
    * 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/storage/StreamBlockInputStream.java

@@ -22,6 +22,7 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -70,8 +71,10 @@ public class StreamBlockInputStream extends BlockExtendedInputStream {
   private final String name = "stream" + STREAM_ID.getAndIncrement();
   private final BlockID blockID;
   private final long blockLength;
-  private final int responseDataSize = 1 << 20; // 1 MB
-  private final long preReadSize = 32 << 20; // 32 MB
+  private final int responseDataSize;
+  private final long preReadSize;
+  private final Duration readTimeout;
+  private final long readTimeoutNanos;
   private final AtomicReference<Pipeline> pipelineRef = new AtomicReference<>();
   private final AtomicReference<Token<OzoneBlockTokenIdentifier>> tokenRef = new AtomicReference<>();
   private XceiverClientFactory xceiverClientFactory;
@@ -101,6 +104,10 @@ public StreamBlockInputStream(
     this.verifyChecksum = config.isChecksumVerify();
     this.retryPolicy = getReadRetryPolicy(config);
     this.refreshFunction = refreshFunction;
+    this.preReadSize = config.getStreamReadPreReadSize();
+    this.responseDataSize = config.getStreamReadResponseDataSize();
+    this.readTimeout = config.getStreamReadTimeout();
+    this.readTimeoutNanos = readTimeout.toNanos();
   }
 
   @Override
@@ -328,6 +335,19 @@ public String toString() {
     return name;
   }
 
+  public long getPreReadSize() {
+    return preReadSize;
+  }
+
+  public int getResponseDataSize() {
+    return responseDataSize;
+  }
+
+  /** Visible for testing: returns the configured streaming read timeout. */
+  public Duration getReadTimeout() {
+    return readTimeout;
+  }
+
   /**
    * Implementation of a StreamObserver used to received and buffer streaming GRPC reads.
    */
@@ -351,10 +371,9 @@ void checkError() throws IOException {
       }
     }
 
-    ReadBlockResponseProto poll(int timeout, TimeUnit timeoutUnit) throws IOException {
-      final long timeoutNanos = timeoutUnit.toNanos(timeout);
+    ReadBlockResponseProto poll() throws IOException {
       final long startTime = System.nanoTime();
-      final long pollTimeoutNanos = Math.min(timeoutNanos / 10, 100_000_000);
+      final long pollTimeoutNanos = Math.min(readTimeoutNanos / 10, 100_000_000);
 
       while (true) {
         checkError();
@@ -374,9 +393,9 @@ ReadBlockResponseProto poll(int timeout, TimeUnit timeoutUnit) throws IOExceptio
         }
 
         final long elapsedNanos = System.nanoTime() - startTime;
-        if (elapsedNanos >= timeoutNanos) {
+        if (elapsedNanos >= readTimeoutNanos) {
           setFailedAndThrow(new TimeoutIOException(
-              "Timed out " + timeout + " " + timeoutUnit + " waiting for response"));
+              "Timed out waiting for response after " + readTimeout));
           return null;
         }
       }
@@ -399,7 +418,7 @@ private ByteBuffer read(int length, boolean preRead) throws IOException {
     }
 
     ByteBuffer readFromQueue() throws IOException {
-      final ReadBlockResponseProto readBlock = poll(10, TimeUnit.SECONDS);
+      final ReadBlockResponseProto readBlock = poll();
       // The server always returns data starting from the last checksum boundary. Therefore if the reader position is
       // ahead of the position we received from the server, we need to adjust the buffer position accordingly.
       // If the reader position is behind

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

@@ -22,6 +22,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.time.Duration;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.junit.jupiter.api.Test;
 
@@ -76,4 +77,18 @@ void testClientHBaseEnhancementsAllowedFalse() {
     assertFalse(subject.getEnablePutblockPiggybacking());
     assertEquals(1, subject.getMaxConcurrentWritePerKey());
   }
+
+  @Test
+  public void testStreamReadConfigParsing() {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.set("ozone.client.stream.read.pre-read-size", "67108864");
+    conf.set("ozone.client.stream.read.response-data-size", "2097152");
+    conf.set("ozone.client.stream.read.timeout", "5s");
+
+    OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+
+    assertEquals(64L << 20, clientConfig.getStreamReadPreReadSize());
+    assertEquals(2 << 20, clientConfig.getStreamReadResponseDataSize());
+    assertEquals(Duration.ofSeconds(5), clientConfig.getStreamReadTimeout());
+  }
 }

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

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.storage;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import java.time.Duration;
+import java.util.function.Function;
+import org.apache.hadoop.hdds.client.BlockID;
+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.security.token.OzoneBlockTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for StreamBlockInputStream custom configuration behavior.
+ */
+public class TestStreamBlockInputStream {
+
+  @Test
+  public void testCustomStreamReadConfigIsApplied() throws Exception {
+    // Arrange: create a config with non-default values
+    OzoneClientConfig clientConfig = new OzoneClientConfig();
+    clientConfig.setStreamReadPreReadSize(64L << 20);
+    clientConfig.setStreamReadResponseDataSize(2 << 20);
+    clientConfig.setStreamReadTimeout(Duration.ofSeconds(5));
+
+    // Sanity check
+    assertEquals(Duration.ofSeconds(5), clientConfig.getStreamReadTimeout());
+    // Create a dummy BlockID for the test
+    BlockID blockID = new BlockID(1L, 1L);
+    long length = 1024L;
+    // Create a mock Pipeline instance.
+    Pipeline pipeline = mock(Pipeline.class);
+
+    Token<OzoneBlockTokenIdentifier> token = null;
+    // Mock XceiverClientFactory since StreamBlockInputStream requires it in the constructor
+    XceiverClientFactory xceiverClientFactory = mock(XceiverClientFactory.class);
+    Function<BlockID, BlockLocationInfo> refreshFunction = b -> null;
+    // Create a StreamBlockInputStream instance
+    try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+        blockID, length, pipeline, token,
+        xceiverClientFactory, refreshFunction, clientConfig)) {
+
+      // Assert: fields should match config values
+      assertEquals(64L << 20, sbis.getPreReadSize());
+      assertEquals(2 << 20, sbis.getResponseDataSize());
+      assertEquals(Duration.ofSeconds(5), sbis.getReadTimeout());
+    }
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -31,6 +31,7 @@
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_PORT_DEFAULT;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CLIENT_PORT_KEY;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_PORT_DEFAULT;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_PORT_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEFAULT_SERVICE_ID;
@@ -57,10 +58,12 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalInt;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
 import javax.management.ObjectName;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.ConfigRedactor;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
@@ -331,6 +334,38 @@ public static Collection<InetSocketAddress> getSCMAddressForDatanodes(
     }
   }
 
+  /**
+   * Returns the SCM address for datanodes based on the service ID and the SCM addresses.
+   * @param conf Configuration
+   * @param scmServiceId SCM service ID
+   * @param scmNodeIds Requested SCM node IDs
+   * @return A collection with addresses of the request SCM node IDs.
+   * Null if there is any wrongly configured SCM address. Note that the returned collection
+   * might not be ordered the same way as the requested SCM node IDs
+   */
+  public static Collection<Pair<String, InetSocketAddress>> getSCMAddressForDatanodes(
+      ConfigurationSource conf, String scmServiceId, Set<String> scmNodeIds) {
+    Collection<Pair<String, InetSocketAddress>> scmNodeAddress = new HashSet<>(scmNodeIds.size());
+    for (String scmNodeId : scmNodeIds) {
+      String addressKey = ConfUtils.addKeySuffixes(
+          OZONE_SCM_ADDRESS_KEY, scmServiceId, scmNodeId);
+      String scmAddress = conf.get(addressKey);
+      if (scmAddress == null) {
+        LOG.warn("The SCM address configuration {} is not defined, return nothing", addressKey);
+        return null;
+      }
+
+      int scmDatanodePort = SCMNodeInfo.getPort(conf, scmServiceId, scmNodeId,
+          OZONE_SCM_DATANODE_ADDRESS_KEY, OZONE_SCM_DATANODE_PORT_KEY,
+          OZONE_SCM_DATANODE_PORT_DEFAULT);
+
+      String scmDatanodeAddressStr = SCMNodeInfo.buildAddress(scmAddress, scmDatanodePort);
+      InetSocketAddress scmDatanodeAddress = NetUtils.createSocketAddr(scmDatanodeAddressStr);
+      scmNodeAddress.add(Pair.of(scmNodeId, scmDatanodeAddress));
+    }
+    return scmNodeAddress;
+  }
+
   /**
    * Retrieve the socket addresses of recon.
    *