HDDS-15076. Fix Incorrect pending deletion size for EC blocks in DN (#10105)

Author: priyeshkaratha

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java

@@ -649,16 +649,16 @@ private void updateMetaData(KeyValueContainerData containerData,
       // Update pending deletion blocks count, blocks bytes and delete transaction ID in in-memory container status.
       // Persist pending bytes only if the feature is finalized.
       if (VersionedDatanodeFeatures.isFinalized(
-          HDDSLayoutFeature.STORAGE_SPACE_DISTRIBUTION) && delTX.hasTotalBlockSize()) {
+          HDDSLayoutFeature.STORAGE_SPACE_DISTRIBUTION) && delTX.hasTotalSizePerReplica()) {
         long pendingBytes = containerData.getBlockPendingDeletionBytes();
-        pendingBytes += delTX.getTotalBlockSize();
+        pendingBytes += delTX.getTotalSizePerReplica();
         metadataTable
             .putWithBatch(batchOperation,
                 containerData.getPendingDeleteBlockBytesKey(),
                 pendingBytes);
       }
       containerData.incrPendingDeletionBlocks(newDeletionBlocks,
-          delTX.hasTotalBlockSize() ? delTX.getTotalBlockSize() : 0);
+          delTX.hasTotalSizePerReplica() ? delTX.getTotalSizePerReplica() : 0);
       containerData.updateDeleteTransactionId(delTX.getTxID());
     }
   }

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

@@ -50,6 +50,7 @@ public KeyBlocks getProto() {
       kbb.addBlocks(deletedBlock.getBlockID().getProtobuf());
       kbb.addSize(deletedBlock.getSize());
       kbb.addReplicatedSize(deletedBlock.getReplicatedSize());
+      kbb.addSizePerReplica(deletedBlock.getSizePerReplica());
     }
     return kbb.setKey(groupID).build();
   }
@@ -62,17 +63,21 @@ public KeyBlocks getProto() {
   public static BlockGroup getFromProto(KeyBlocks proto) {
     List<DeletedBlock> deletedBlocksList = new ArrayList<>();
     for (int i = 0; i < proto.getBlocksCount(); i++) {
-      long repSize = SIZE_NOT_AVAILABLE;
       long size = SIZE_NOT_AVAILABLE;
+      long repSize = SIZE_NOT_AVAILABLE;
+      long sizePerReplica = SIZE_NOT_AVAILABLE;
       if (proto.getSizeCount() > i) {
         size = proto.getSize(i);
       }
       if (proto.getReplicatedSizeCount() > i) {
         repSize = proto.getReplicatedSize(i);
       }
+      if (proto.getSizePerReplicaCount() > i) {
+        sizePerReplica = proto.getSizePerReplica(i);
+      }
       BlockID block = new BlockID(proto.getBlocks(i).getContainerBlockID().getContainerID(),
           proto.getBlocks(i).getContainerBlockID().getLocalID());
-      deletedBlocksList.add(new DeletedBlock(block, size, repSize));
+      deletedBlocksList.add(new DeletedBlock(block, size, repSize, sizePerReplica));
     }
     return BlockGroup.newBuilder().setKeyName(proto.getKey())
         .addAllDeletedBlocks(deletedBlocksList)

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

@@ -27,11 +27,13 @@ public class DeletedBlock {
   private BlockID blockID;
   private long size;
   private long replicatedSize;
+  private long sizePerReplica;
 
-  public DeletedBlock(BlockID blockID, long size, long replicatedSize) {
+  public DeletedBlock(BlockID blockID, long size, long replicatedSize, long sizePerReplica) {
     this.blockID = blockID;
     this.size = size;
     this.replicatedSize = replicatedSize;
+    this.sizePerReplica = sizePerReplica;
   }
 
   public BlockID getBlockID() {
@@ -46,13 +48,18 @@ public long getReplicatedSize() {
     return this.replicatedSize;
   }
 
+  public long getSizePerReplica() {
+    return this.sizePerReplica;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder(64);
     sb.append(" localID: ").append(blockID.getContainerBlockID().getLocalID())
         .append(" containerID: ").append(blockID.getContainerBlockID().getContainerID())
         .append(" size: ").append(size)
-        .append(" replicatedSize: ").append(replicatedSize);
+        .append(" replicatedSize: ").append(replicatedSize)
+        .append(" sizePerReplica: ").append(sizePerReplica);
     return sb.toString();
   }
 }

hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto

@@ -377,6 +377,7 @@ message DeletedBlocksTransaction {
   optional int32 count = 4 [deprecated=true];
   optional uint64 totalBlockSize = 5;
   optional uint64 totalBlockReplicatedSize = 6;
+  optional uint64 totalSizePerReplica = 7;
 }
 
 // ACK message datanode sent to SCM, contains the result of

hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto

@@ -183,6 +183,7 @@ message KeyBlocks {
   repeated BlockID blocks = 2;
   repeated uint64 size = 3;
   repeated uint64 replicatedSize = 4;
+  repeated uint64 sizePerReplica = 5;
 }
 
 /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImpl.java

@@ -171,6 +171,11 @@ private DeletedBlocksTransaction constructNewTransaction(
         builder.setTotalBlockReplicatedSize(replicatedSize);
         builder.setTotalBlockSize(blocks.stream().mapToLong(DeletedBlock::getSize).sum());
       }
+
+      long sizePerReplica = blocks.stream().mapToLong(DeletedBlock::getSizePerReplica).sum();
+      if (sizePerReplica >= 0) {
+        builder.setTotalSizePerReplica(sizePerReplica);
+      }
     }
     return builder.build();
   }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/block/TestDeletedBlockLog.java

@@ -231,7 +231,8 @@ private Map<Long, List<DeletedBlock>> generateData(int txCount,
       updateContainerMetadata(containerID, state);
       for (int j = 0; j < BLOCKS_PER_TXN; j++)  {
         long localID = localIDBase + j;
-        blocks.add(new DeletedBlock(new BlockID(containerID, localID), blockSize + j, blockSize + j));
+        blocks.add(new DeletedBlock(new BlockID(containerID, localID),
+            blockSize + j, blockSize + j, blockSize + j));
       }
       blockMap.put(containerID, blocks);
     }
@@ -733,7 +734,8 @@ public void testDeletedBlockTransactions() throws IOException {
     Map<Long, List<DeletedBlock>> deletedBlocksMap = new HashMap<>();
     long localId = RandomUtils.secure().randomLong();
     List<DeletedBlock> blockIDList = new ArrayList<>();
-    blockIDList.add(new DeletedBlock(new BlockID(containerID, localId), SIZE_NOT_AVAILABLE, SIZE_NOT_AVAILABLE));
+    blockIDList.add(new DeletedBlock(new BlockID(containerID, localId),
+        SIZE_NOT_AVAILABLE, SIZE_NOT_AVAILABLE, SIZE_NOT_AVAILABLE));
     deletedBlocksMap.put(containerID, blockIDList);
     addTransactions(deletedBlocksMap, true);
     blocks = getTransactions(txNum * BLOCKS_PER_TXN * ONE);
@@ -808,7 +810,8 @@ public void testTransactionSerializedSize(int blockCount) {
     long containerID = 1000000;
     List<DeletedBlock> blocks = new ArrayList<>();
     for (int i = 0; i < blockCount; i++) {
-      blocks.add(new DeletedBlock(new BlockID(containerID, 100000000 + i), 128 * 1024 * 1024, 128 * 1024 * 1024));
+      blocks.add(new DeletedBlock(new BlockID(containerID, 100000000 + i),
+          128 * 1024 * 1024, 128 * 1024 * 1024, 128 * 1024 * 1024));
     }
     List<Long> localIdList = blocks.stream().map(b -> b.getBlockID().getLocalID()).collect(Collectors.toList());
     DeletedBlocksTransaction tx1 = DeletedBlocksTransaction.newBuilder()
@@ -818,6 +821,7 @@ public void testTransactionSerializedSize(int blockCount) {
         .setCount(0)
         .setTotalBlockSize(blocks.stream().mapToLong(DeletedBlock::getSize).sum())
         .setTotalBlockReplicatedSize(blocks.stream().mapToLong(DeletedBlock::getReplicatedSize).sum())
+        .setTotalSizePerReplica(blocks.stream().mapToLong(DeletedBlock::getSizePerReplica).sum())
         .build();
     DeletedBlocksTransaction tx2 = DeletedBlocksTransaction.newBuilder()
         .setTxID(txID)

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/QuotaUtil.java

@@ -66,6 +66,27 @@ public static long getReplicatedSize(
     }
   }
 
+  /**
+   * Calculate the physical bytes stored on a single DataNode for one block.
+   * For RATIS each DN holds a full copy, so the per-replica size equals the
+   * logical block size. For EC each DN holds one chunk per stripe, so the
+   * per-replica size is {@code replicatedSize / requiredNodes}.
+   *
+   * @param dataSize  logical (unreplicated) block size in bytes
+   * @param repConfig replication configuration
+   * @return bytes stored on a single DN for this block
+   */
+  public static long getSizePerReplica(long dataSize, ReplicationConfig repConfig) {
+    if (repConfig.getReplicationType() == RATIS) {
+      return dataSize;
+    }
+    int requiredNodes = repConfig.getRequiredNodes();
+    if (requiredNodes <= 0) {
+      return dataSize;
+    }
+    return getReplicatedSize(dataSize, repConfig) / requiredNodes;
+  }
+
   /**
    * Get an estimated data size (before replication) from the replicated size.
    * An (inaccurate) reverse of getReplicatedSize().

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestQuotaUtil.java

@@ -97,4 +97,28 @@ public void testECPartialSingleBeyondFirstChunk() {
     assertEquals(dataSize + ONE_MB * 2, replicatedSize);
   }
 
+  @Test
+  public void testGetSizePerReplicaRatisThree() {
+    ReplicationConfig repConfig = RatisReplicationConfig.getInstance(THREE);
+    // Each of the 3 DNs holds the full block, so per-replica == logical size.
+    assertEquals(123 * ONE_MB, QuotaUtil.getSizePerReplica(123 * ONE_MB, repConfig));
+  }
+
+  @Test
+  public void testGetSizePerReplicaRatisOne() {
+    ReplicationConfig repConfig = RatisReplicationConfig.getInstance(ONE);
+    assertEquals(123 * ONE_MB, QuotaUtil.getSizePerReplica(123 * ONE_MB, repConfig));
+  }
+
+  @Test
+  public void testGetSizePerReplicaEC() {
+    // EC(3,2): 5 nodes total. replicatedSize = dataSize + parity overhead.
+    // Per-replica = replicatedSize / 5.
+    ECReplicationConfig repConfig = new ECReplicationConfig(3, 2, RS, ONE_MB);
+    long dataSize = ONE_MB * 3 * 10; // 10 full stripes
+    long replicatedSize = QuotaUtil.getReplicatedSize(dataSize, repConfig);
+    long expected = replicatedSize / repConfig.getRequiredNodes(); // / 5
+    assertEquals(expected, QuotaUtil.getSizePerReplica(dataSize, repConfig));
+  }
+
 }