Hdds 13184 persist block size

dev-support/pom.xml

@@ -17,7 +17,7 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>ozone-main</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
   </parent>
   <artifactId>ozone-dev-support</artifactId>
   <name>Apache Ozone Dev Support</name>

hadoop-hdds/annotations/pom.xml

@@ -17,11 +17,11 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>hdds</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>hdds-annotation-processing</artifactId>
-  <version>2.1.0-SNAPSHOT</version>
+  <version>2.2.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Apache Ozone Annotation Processing</name>
   <description>Apache Ozone annotation processing tools for validating custom

hadoop-hdds/client/pom.xml

@@ -17,12 +17,12 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>hdds-hadoop-dependency-client</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
     <relativePath>../hadoop-dependency-client</relativePath>
   </parent>
 
   <artifactId>hdds-client</artifactId>
-  <version>2.1.0-SNAPSHOT</version>
+  <version>2.2.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Apache Ozone HDDS Client</name>
   <description>Apache Ozone Distributed Data Store Client Library</description>

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

@@ -0,0 +1,148 @@
+/*
+ * 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.ozone.client.io;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ComparisonChain;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.io.ByteBufferPool;
+
+/**
+ * A bounded version of ElasticByteBufferPool that limits the total size
+ * of buffers that can be cached in the pool. This prevents unbounded memory
+ * growth in long-lived rpc clients like S3 Gateway.
+ *
+ * When the pool reaches its maximum size, newly returned buffers are not
+ * added back to the pool and will be garbage collected instead.
+ */
+public class BoundedElasticByteBufferPool implements ByteBufferPool {
+  private final TreeMap<Key, ByteBuffer> buffers = new TreeMap<>();
+  private final TreeMap<Key, ByteBuffer> directBuffers = new TreeMap<>();
+  private final long maxPoolSize;
+  private  final AtomicLong currentPoolSize = new AtomicLong(0);
+
+  /**
+   * A logical timestamp counter used for creating unique Keys in the TreeMap.
+   * This is used as the insertionTime for the Key instead of System.nanoTime()
+   * to guarantee uniqueness and avoid a potential spin-wait in putBuffer
+   * if two buffers of the same capacity are added at the same nanosecond.
+   */
+  private long logicalTimestamp = 0;
+
+  public BoundedElasticByteBufferPool(long maxPoolSize) {
+    super();
+    this.maxPoolSize = maxPoolSize;
+  }
+
+  private TreeMap<Key, ByteBuffer> getBufferTree(boolean direct) {
+    return direct ? this.directBuffers : this.buffers;
+  }
+
+  @Override
+  public synchronized ByteBuffer getBuffer(boolean direct, int length) {
+    TreeMap<Key, ByteBuffer> tree = this.getBufferTree(direct);
+    Map.Entry<Key, ByteBuffer> entry = tree.ceilingEntry(new Key(length, 0L));
+    if (entry == null) {
+      // Pool is empty or has no suitable buffer. Allocate a new one.
+      return direct ? ByteBuffer.allocateDirect(length) : ByteBuffer.allocate(length);
+    }
+    tree.remove(entry.getKey());
+    ByteBuffer buffer = entry.getValue();
+
+    // Decrement the size because we are taking a buffer OUT of the pool.
+    currentPoolSize.addAndGet(-buffer.capacity());
+    buffer.clear();
+    return buffer;
+  }
+
+  @Override
+  public synchronized void putBuffer(ByteBuffer buffer) {
+    if (buffer == null) {
+      return;
+    }
+
+    if (currentPoolSize.get() + buffer.capacity() > maxPoolSize) {
+      // Pool is full, do not add the buffer back.
+      // It will be garbage collected by JVM.
+      return;
+    }
+
+    buffer.clear();
+    TreeMap<Key, ByteBuffer> tree = getBufferTree(buffer.isDirect());
+    Key key = new Key(buffer.capacity(), logicalTimestamp++);
+
+    tree.put(key, buffer);
+    // Increment the size because we have successfully added buffer back to the pool.
+    currentPoolSize.addAndGet(buffer.capacity());
+  }
+
+  /**
+   * Get the current size of buffers in the pool.
+   *
+   * @return Current pool size in bytes
+   */
+  @VisibleForTesting
+  public synchronized long getCurrentPoolSize() {
+    return currentPoolSize.get();
+  }
+
+  /**
+   * The Key for the buffer TreeMaps.
+   * This is copied directly from the original ElasticByteBufferPool.
+   */
+  protected static final class Key implements Comparable<Key> {
+    private final int capacity;
+    private final long insertionTime;
+
+    Key(int capacity, long insertionTime) {
+      this.capacity = capacity;
+      this.insertionTime = insertionTime;
+    }
+
+    @Override
+    public int compareTo(Key other) {
+      return ComparisonChain.start()
+          .compare(this.capacity, other.capacity)
+          .compare(this.insertionTime, other.insertionTime)
+          .result();
+    }
+
+    @Override
+    public boolean equals(Object rhs) {
+      if (rhs == null) {
+        return false;
+      }
+      try {
+        Key o = (Key) rhs;
+        return compareTo(o) == 0;
+      } catch (ClassCastException e) {
+        return false;
+      }
+    }
+
+    @Override
+    public int hashCode() {
+      return new HashCodeBuilder().append(capacity).append(insertionTime)
+          .toHashCode();
+    }
+  }
+}

hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestBoundedElasticByteBufferPool.java

@@ -0,0 +1,121 @@
+/*
+ * 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.ozone.client.io;
+
+import java.nio.ByteBuffer;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for BoundedElasticByteBufferPool.
+ */
+public class TestBoundedElasticByteBufferPool {
+
+  private static final int MB = 1024 * 1024;
+  private static final long MAX_POOL_SIZE = 3L * MB; // 3MB
+
+  @Test
+  public void testLogicalTimestampOrdering() {
+    // Pool with plenty of capacity
+    BoundedElasticByteBufferPool pool = new BoundedElasticByteBufferPool(MAX_POOL_SIZE);
+    int bufferSize = 5 * 1024; // 5KB
+
+    // Create and add three distinct buffers of the same size
+    ByteBuffer buffer1 = ByteBuffer.allocate(bufferSize);
+    ByteBuffer buffer2 = ByteBuffer.allocate(bufferSize);
+    ByteBuffer buffer3 = ByteBuffer.allocate(bufferSize);
+
+    // Store their unique identity hash codes
+    int hash1 = System.identityHashCode(buffer1);
+    int hash2 = System.identityHashCode(buffer2);
+    int hash3 = System.identityHashCode(buffer3);
+
+    pool.putBuffer(buffer1);
+    pool.putBuffer(buffer2);
+    pool.putBuffer(buffer3);
+
+    // The pool should now contain 15KB data
+    Assertions.assertEquals(bufferSize * 3L, pool.getCurrentPoolSize());
+
+    // Get the buffers back. They should come back in the same
+    // order they were put in (FIFO).
+    ByteBuffer retrieved1 = pool.getBuffer(false, bufferSize);
+    ByteBuffer retrieved2 = pool.getBuffer(false, bufferSize);
+    ByteBuffer retrieved3 = pool.getBuffer(false, bufferSize);
+
+    // Verify we got the exact same buffer instances back in FIFO order
+    Assertions.assertEquals(hash1, System.identityHashCode(retrieved1));
+    Assertions.assertEquals(hash2, System.identityHashCode(retrieved2));
+    Assertions.assertEquals(hash3, System.identityHashCode(retrieved3));
+
+    // The pool should now be empty
+    Assertions.assertEquals(0, pool.getCurrentPoolSize());
+  }
+
+  /**
+   * Verifies the core feature: the pool stops caching buffers
+   * once its maximum size is reached.
+   */
+  @Test
+  public void testPoolBoundingLogic() {
+    BoundedElasticByteBufferPool pool = new BoundedElasticByteBufferPool(MAX_POOL_SIZE);
+
+    ByteBuffer buffer1 = ByteBuffer.allocate(2 * MB);
+    ByteBuffer buffer2 = ByteBuffer.allocate(1 * MB);
+    ByteBuffer buffer3 = ByteBuffer.allocate(3 * MB);
+
+    int hash1 = System.identityHashCode(buffer1);
+    int hash2 = System.identityHashCode(buffer2);
+    int hash3 = System.identityHashCode(buffer3);
+
+    // 1. Put buffer 1 (Pool size: 2MB, remaining: 1MB)
+    pool.putBuffer(buffer1);
+    Assertions.assertEquals(2 * MB, pool.getCurrentPoolSize());
+
+    // 2. Put buffer 2 (Pool size: 2MB + 1MB = 3MB, remaining: 0)
+    // The check is (current(2MB) + new(1MB)) > max(3MB), which is false.
+    // So, the buffer IS added.
+    pool.putBuffer(buffer2);
+    Assertions.assertEquals(3 * MB, pool.getCurrentPoolSize());
+
+    // 3. Put buffer 3 (Capacity 3MB)
+    // The check is (current(3MB) + new(3MB)) > max(3MB), which is true.
+    // This buffer should be REJECTED.
+    pool.putBuffer(buffer3);
+    // The pool size should NOT change.
+    Assertions.assertEquals(3 * MB, pool.getCurrentPoolSize());
+
+    // 4. Get buffers back
+    ByteBuffer retrieved1 = pool.getBuffer(false, 2 * MB);
+    ByteBuffer retrieved2 = pool.getBuffer(false, 1 * MB);
+
+    // The pool should now be empty
+    Assertions.assertEquals(0, pool.getCurrentPoolSize());
+
+    // 5. Ask for a third buffer.
+    // Since buffer3 was rejected, this should be a NEWLY allocated buffer.
+    ByteBuffer retrieved3 = pool.getBuffer(false, 3 * MB);
+
+    // Verify that we got the first two buffers from the pool
+    Assertions.assertEquals(hash1, System.identityHashCode(retrieved1));
+    Assertions.assertEquals(hash2, System.identityHashCode(retrieved2));
+
+    // Verify that the third buffer is a NEW instance, not buffer3
+    Assertions.assertNotEquals(hash3, System.identityHashCode(retrieved3));
+  }
+}

hadoop-hdds/common/pom.xml

@@ -17,11 +17,11 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>hdds-hadoop-dependency-client</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
     <relativePath>../hadoop-dependency-client</relativePath>
   </parent>
   <artifactId>hdds-common</artifactId>
-  <version>2.1.0-SNAPSHOT</version>
+  <version>2.2.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Apache Ozone HDDS Common</name>
   <description>Apache Ozone Distributed Data Store Common</description>

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/upgrade/HDDSLayoutFeature.java

@@ -42,7 +42,8 @@ public enum HDDSLayoutFeature implements LayoutFeature {
                                      "to DatanodeDetails."),
   HBASE_SUPPORT(8, "Datanode RocksDB Schema Version 3 has an extra table " +
           "for the last chunk of blocks to support HBase.)"),
-  WITNESSED_CONTAINER_DB_PROTO_VALUE(9, "ContainerID table schema to use value type as proto");
+  WITNESSED_CONTAINER_DB_PROTO_VALUE(9, "ContainerID table schema to use value type as proto"),
+  STORAGE_DATA_DISTRIBUTION(10, "ContainerID table schema to use value type as proto");
 
   //////////////////////////////  //////////////////////////////
 

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -690,6 +690,10 @@ public final class OzoneConfigKeys {
       "ozone.security.crypto.compliance.mode";
   public static final String OZONE_SECURITY_CRYPTO_COMPLIANCE_MODE_UNRESTRICTED = "unrestricted";
 
+  public static final String OZONE_CLIENT_ELASTIC_BYTE_BUFFER_POOL_MAX_SIZE =
+      "ozone.client.elastic.byte.buffer.pool.max.size";
+  public static final String OZONE_CLIENT_ELASTIC_BYTE_BUFFER_POOL_MAX_SIZE_DEFAULT = "16GB";
+
   /**
    * There is no need to instantiate this class.
    */

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -221,6 +221,7 @@ public final class OzoneConsts {
   public static final String OM_SST_FILE_INFO_START_KEY = "startKey";
   public static final String OM_SST_FILE_INFO_END_KEY = "endKey";
   public static final String OM_SST_FILE_INFO_COL_FAMILY = "columnFamily";
+  public static final String OM_SLD_TXN_INFO = "transactionInfo";
 
   // YAML fields for .container files
   public static final String CONTAINER_ID = "containerID";

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -465,6 +465,18 @@
     <description>Socket timeout for Ozone client. Unit could be defined with
       postfix (ns,ms,s,m,h,d)</description>
   </property>
+  <property>
+    <name>ozone.client.elastic.byte.buffer.pool.max.size</name>
+    <value>16GB</value>
+    <tag>OZONE, CLIENT</tag>
+    <description>
+      The maximum total size of buffers that can be cached in the client-side
+      ByteBufferPool. This pool is used heavily during EC read and write operations.
+      Setting a limit prevents unbounded memory growth in long-lived rpc clients
+      like the S3 Gateway. Once this limit is reached, used buffers are not
+      put back to the pool and will be garbage collected.
+    </description>
+  </property>
   <property>
     <name>ozone.key.deleting.limit.per.task</name>
     <value>50000</value>

hadoop-hdds/config/pom.xml

@@ -17,10 +17,10 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>hdds</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
   </parent>
   <artifactId>hdds-config</artifactId>
-  <version>2.1.0-SNAPSHOT</version>
+  <version>2.2.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Apache Ozone HDDS Config</name>
   <description>Apache Ozone Distributed Data Store Config Tools</description>

hadoop-hdds/container-service/pom.xml

@@ -17,10 +17,10 @@
   <parent>
     <groupId>org.apache.ozone</groupId>
     <artifactId>hdds</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0-SNAPSHOT</version>
   </parent>
   <artifactId>hdds-container-service</artifactId>
-  <version>2.1.0-SNAPSHOT</version>
+  <version>2.2.0-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>Apache Ozone HDDS Container Service</name>
   <description>Apache Ozone Distributed Data Store Container Service</description>