HDDS-10431. Merge recent commits from master (7c8160fe) to HDDS-7593

.github/dependabot.yml

@@ -38,5 +38,7 @@ updates:
     directory: "/"
     schedule:
       interval: "weekly"
+      day: "saturday"
+      time: "07:00" # UTC
     pull-request-branch-name:
       separator: "-"

.github/labeler.yml

@@ -0,0 +1,21 @@
+# 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.
+
+# Configuration for .github/workflows/label-pr.yml
+
+# This rule can be deleted once the container reconciliation feature branch is merged.
+container-reconciliation:
+- base-branch: HDDS-10239-container-reconciliation
+

.github/workflows/label-pr.yml

@@ -0,0 +1,29 @@
+# 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.
+
+# This workflow reads its configuration from the .github/labeler.yml file.
+name: pull-request-labeler
+on:
+- pull_request_target
+
+jobs:
+  labeler:
+    permissions:
+      contents: read
+      pull-requests: write
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/labeler@v5
+

SECURITY.md

@@ -5,13 +5,16 @@
 The first stable release of Apache Ozone is 1.0, the previous alpha and beta releases are not supported by the community.
 
 | Version       | Supported          |
-| ------------- | ------------------ |
+|---------------| ------------------ |
 | 0.3.0 (alpha) | :x:                |
 | 0.4.0 (alpha) | :x:                |
 | 0.4.1 (alpha) | :x:                |
 | 0.5.0 (beta)  | :x:                |
-| 1.0           | :white_check_mark: |
-| 1.1           | :white_check_mark: |
+| 1.0.0         | :x:                |
+| 1.1.0         | :x:                |
+| 1.2.1         | :x:                |
+| 1.3.0         | :x:                |
+| 1.4.0         | :white_check_mark: |
 
 ## Reporting a Vulnerability
 

dev-support/ci/selective_ci_checks.bats

@@ -57,6 +57,18 @@ load bats-assert/load.bash
   assert_output -p needs-kubernetes-tests=false
 }
 
+@test "dashboard only" {
+  run dev-support/ci/selective_ci_checks.sh 039dea9
+
+  assert_output -p 'basic-checks=["rat"]'
+  assert_output -p needs-build=false
+  assert_output -p needs-compile=false
+  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 "compose and robot" {
   run dev-support/ci/selective_ci_checks.sh b83039eef
 

dev-support/ci/selective_ci_checks.sh

@@ -233,6 +233,7 @@ function get_count_compose_files() {
     local ignore_array=(
         "^hadoop-ozone/dist/src/main/k8s"
         "^hadoop-ozone/dist/src/main/license"
+        "^hadoop-ozone/dist/src/main/compose/common/grafana/dashboards"
         "\.md$"
     )
     filter_changed_files true
@@ -494,6 +495,7 @@ function get_count_misc_files() {
         "\.md$"
         "findbugsExcludeFile.xml"
         "/NOTICE$"
+        "^hadoop-ozone/dist/src/main/compose/common/grafana/dashboards"
     )
     local ignore_array=(
         "^.github/workflows/post-commit.yml"

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

@@ -111,28 +111,23 @@ public void recordWriteChunk(Pipeline pipeline, long chunkSizeBytes) {
     totalWriteChunkBytes.incr(chunkSizeBytes);
   }
 
-  @VisibleForTesting
-  public MutableCounterLong getTotalWriteChunkBytes() {
+  MutableCounterLong getTotalWriteChunkBytes() {
     return totalWriteChunkBytes;
   }
 
-  @VisibleForTesting
-  public MutableCounterLong getTotalWriteChunkCalls() {
+  MutableCounterLong getTotalWriteChunkCalls() {
     return totalWriteChunkCalls;
   }
 
-  @VisibleForTesting
-  public Map<PipelineID, MutableCounterLong> getWriteChunkBytesByPipeline() {
+  Map<PipelineID, MutableCounterLong> getWriteChunkBytesByPipeline() {
     return writeChunkBytesByPipeline;
   }
 
-  @VisibleForTesting
-  public Map<PipelineID, MutableCounterLong> getWriteChunkCallsByPipeline() {
+  Map<PipelineID, MutableCounterLong> getWriteChunkCallsByPipeline() {
     return writeChunkCallsByPipeline;
   }
 
-  @VisibleForTesting
-  public Map<UUID, MutableCounterLong> getWriteChunksCallsByLeaders() {
+  Map<UUID, MutableCounterLong> getWriteChunksCallsByLeaders() {
     return writeChunksCallsByLeaders;
   }
 }

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

@@ -201,6 +201,13 @@ public enum ChecksumCombineMode {
   // 3 concurrent stripe read should be enough.
   private int ecReconstructStripeReadPoolLimit = 10 * 3;
 
+  @Config(key = "ec.reconstruct.stripe.write.pool.limit",
+      defaultValue = "30",
+      description = "Thread pool max size for parallelly write" +
+          " available ec chunks to reconstruct the whole stripe.",
+      tags = ConfigTag.CLIENT)
+  private int ecReconstructStripeWritePoolLimit = 10 * 3;
+
   @Config(key = "checksum.combine.mode",
       defaultValue = "COMPOSITE_CRC",
       description = "The combined checksum type [MD5MD5CRC / COMPOSITE_CRC] "
@@ -233,7 +240,7 @@ public enum ChecksumCombineMode {
   private boolean incrementalChunkList = true;
 
   @PostConstruct
-  private void validate() {
+  public void validate() {
     Preconditions.checkState(streamBufferSize > 0);
     Preconditions.checkState(streamBufferFlushSize > 0);
     Preconditions.checkState(streamBufferMaxSize > 0);
@@ -396,6 +403,14 @@ public int getEcReconstructStripeReadPoolLimit() {
     return ecReconstructStripeReadPoolLimit;
   }
 
+  public void setEcReconstructStripeWritePoolLimit(int poolLimit) {
+    this.ecReconstructStripeWritePoolLimit = poolLimit;
+  }
+
+  public int getEcReconstructStripeWritePoolLimit() {
+    return ecReconstructStripeWritePoolLimit;
+  }
+
   public void setFsDefaultBucketLayout(String bucketLayout) {
     if (!bucketLayout.isEmpty()) {
       this.fsDefaultBucketLayout = bucketLayout;

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

@@ -166,8 +166,8 @@ private synchronized void connectToDatanode(DatanodeDetails dn)
     // port.
     int port = dn.getPort(DatanodeDetails.Port.Name.STANDALONE).getValue();
     if (port == 0) {
-      port = config.getInt(OzoneConfigKeys.DFS_CONTAINER_IPC_PORT,
-          OzoneConfigKeys.DFS_CONTAINER_IPC_PORT_DEFAULT);
+      port = config.getInt(OzoneConfigKeys.HDDS_CONTAINER_IPC_PORT,
+          OzoneConfigKeys.HDDS_CONTAINER_IPC_PORT_DEFAULT);
     }
 
     // Add credential context to the client call

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

@@ -83,8 +83,8 @@ public static XceiverClientRatis newXceiverClientRatis(
       org.apache.hadoop.hdds.scm.pipeline.Pipeline pipeline,
       ConfigurationSource ozoneConf, ClientTrustManager trustManager) {
     final String rpcType = ozoneConf
-        .get(ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_KEY,
-            ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_DEFAULT);
+        .get(ScmConfigKeys.HDDS_CONTAINER_RATIS_RPC_TYPE_KEY,
+            ScmConfigKeys.HDDS_CONTAINER_RATIS_RPC_TYPE_DEFAULT);
     final RetryPolicy retryPolicy = RatisHelper.createRetryPolicy(ozoneConf);
     final GrpcTlsConfig tlsConfig = RatisHelper.createTlsClientConfig(new
         SecurityConfig(ozoneConf), trustManager);

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

@@ -73,7 +73,7 @@ SortedMap<Long, List<BUFFER>> getCommitIndexMap() {
     return commitIndexMap;
   }
 
-  void updateCommitInfoMap(long index, List<BUFFER> buffers) {
+  synchronized void updateCommitInfoMap(long index, List<BUFFER> buffers) {
     commitIndexMap.computeIfAbsent(index, k -> new LinkedList<>())
         .addAll(buffers);
   }

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

@@ -27,9 +27,9 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -157,7 +157,8 @@ public BlockOutputStream(
       BufferPool bufferPool,
       OzoneClientConfig config,
       Token<? extends TokenIdentifier> token,
-      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs
+      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs,
+      Supplier<ExecutorService> blockOutputStreamResourceProvider
   ) throws IOException {
     this.xceiverClientFactory = xceiverClientManager;
     this.config = config;
@@ -199,8 +200,7 @@ public BlockOutputStream(
             (long) flushPeriod * streamBufferArgs.getStreamBufferSize() == streamBufferArgs
                 .getStreamBufferFlushSize());
 
-    // A single thread executor handle the responses of async requests
-    responseExecutor = Executors.newSingleThreadExecutor();
+    this.responseExecutor = blockOutputStreamResourceProvider.get();
     bufferList = null;
     totalDataFlushedLength = 0;
     writtenDataLength = 0;
@@ -682,7 +682,6 @@ public void cleanup(boolean invalidateClient) {
       bufferList.clear();
     }
     bufferList = null;
-    responseExecutor.shutdown();
   }
 
   /**

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

@@ -111,6 +111,7 @@ void releaseBuffer(ChunkBuffer chunkBuffer) {
   }
 
   public void clearBufferPool() {
+    bufferList.forEach(ChunkBuffer::close);
     bufferList.clear();
     currentBufferIndex = -1;
   }

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

@@ -24,6 +24,7 @@
  */
 package org.apache.hadoop.hdds.scm.storage;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.ozone.common.ChunkBuffer;
@@ -32,6 +33,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
 
 /**
  * This class executes watchForCommit on ratis pipeline and releases
@@ -42,8 +44,8 @@ class CommitWatcher extends AbstractCommitWatcher<ChunkBuffer> {
   private final BufferPool bufferPool;
 
   // future Map to hold up all putBlock futures
-  private final ConcurrentMap<Long, CompletableFuture<
-      ContainerCommandResponseProto>> futureMap = new ConcurrentHashMap<>();
+  private final ConcurrentMap<Long, CompletableFuture<ContainerCommandResponseProto>>
+      futureMap = new ConcurrentHashMap<>();
 
   CommitWatcher(BufferPool bufferPool, XceiverClientSpi xceiverClient) {
     super(xceiverClient);
@@ -67,11 +69,24 @@ void releaseBuffers(long index) {
         + totalLength + ": existing = " + futureMap.keySet());
   }
 
-  ConcurrentMap<Long, CompletableFuture<
-      ContainerCommandResponseProto>> getFutureMap() {
+  @VisibleForTesting
+  ConcurrentMap<Long, CompletableFuture<ContainerCommandResponseProto>> getFutureMap() {
     return futureMap;
   }
 
+  public void putFlushFuture(long flushPos, CompletableFuture<ContainerCommandResponseProto> flushFuture) {
+    futureMap.compute(flushPos,
+        (key, previous) -> previous == null ? flushFuture :
+            previous.thenCombine(flushFuture, (prev, curr) -> curr));
+  }
+
+
+  public void waitOnFlushFutures() throws InterruptedException, ExecutionException {
+    // wait for all the transactions to complete
+    CompletableFuture.allOf(futureMap.values().toArray(
+        new CompletableFuture[0])).get();
+  }
+
   @Override
   public void cleanup() {
     super.cleanup();

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

@@ -44,6 +44,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.putBlockAsync;
@@ -75,10 +77,11 @@ public ECBlockOutputStream(
       BufferPool bufferPool,
       OzoneClientConfig config,
       Token<? extends TokenIdentifier> token,
-      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs
+      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs,
+      Supplier<ExecutorService> executorServiceSupplier
   ) throws IOException {
     super(blockID, xceiverClientManager,
-        pipeline, bufferPool, config, token, clientMetrics, streamBufferArgs);
+        pipeline, bufferPool, config, token, clientMetrics, streamBufferArgs, executorServiceSupplier);
     // In EC stream, there will be only one node in pipeline.
     this.datanodeDetails = pipeline.getClosestNode();
   }

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

@@ -37,6 +37,8 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 
 /**
  * An {@link OutputStream} used by the REST service in combination with the
@@ -65,8 +67,8 @@ public class RatisBlockOutputStream extends BlockOutputStream
   /**
    * Creates a new BlockOutputStream.
    *
-   * @param blockID              block ID
-   * @param bufferPool           pool of buffers
+   * @param blockID    block ID
+   * @param bufferPool pool of buffers
    */
   @SuppressWarnings("checkstyle:ParameterNumber")
   public RatisBlockOutputStream(
@@ -76,10 +78,11 @@ public RatisBlockOutputStream(
       BufferPool bufferPool,
       OzoneClientConfig config,
       Token<? extends TokenIdentifier> token,
-      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs
+      ContainerClientMetrics clientMetrics, StreamBufferArgs streamBufferArgs,
+      Supplier<ExecutorService> blockOutputStreamResourceProvider
   ) throws IOException {
     super(blockID, xceiverClientManager, pipeline,
-        bufferPool, config, token, clientMetrics, streamBufferArgs);
+        bufferPool, config, token, clientMetrics, streamBufferArgs, blockOutputStreamResourceProvider);
     this.commitWatcher = new CommitWatcher(bufferPool, getXceiverClient());
   }
 
@@ -110,16 +113,13 @@ void updateCommitInfo(XceiverClientReply reply, List<ChunkBuffer> buffers) {
   }
 
   @Override
-  void putFlushFuture(long flushPos,
-      CompletableFuture<ContainerCommandResponseProto> flushFuture) {
-    commitWatcher.getFutureMap().put(flushPos, flushFuture);
+  void putFlushFuture(long flushPos, CompletableFuture<ContainerCommandResponseProto> flushFuture) {
+    commitWatcher.putFlushFuture(flushPos, flushFuture);
   }
 
   @Override
   void waitOnFlushFutures() throws InterruptedException, ExecutionException {
-    // wait for all the transactions to complete
-    CompletableFuture.allOf(commitWatcher.getFutureMap().values().toArray(
-        new CompletableFuture[0])).get();
+    commitWatcher.waitOnFlushFutures();
   }
 
   @Override