public void updateQuotaAndCollectBlocks(INode.ReclaimContext reclaimContext,
      INodeFile file, FileDiff removed) {
    byte storagePolicyID = file.getStoragePolicyID();
    BlockStoragePolicy bsp = null;
    if (storagePolicyID != HdfsConstants.BLOCK_STORAGE_POLICY_ID_UNSPECIFIED) {
      bsp = reclaimContext.storagePolicySuite().
          getPolicy(file.getStoragePolicyID());
    }

    QuotaCounts oldCounts;
    if (removed.snapshotINode != null) {
      oldCounts = new QuotaCounts.Builder().build();
      // collect all distinct blocks
      Set<BlockInfo> allBlocks = new HashSet<BlockInfo>();
      if (file.getBlocks() != null) {
        allBlocks.addAll(Arrays.asList(file.getBlocks()));
      }
      if (removed.getBlocks() != null) {
        allBlocks.addAll(Arrays.asList(removed.getBlocks()));
      }
      for (FileDiff diff : diffs) {
        BlockInfo[] diffBlocks = diff.getBlocks();
        if (diffBlocks != null) {
          allBlocks.addAll(Arrays.asList(diffBlocks));
        }
      }
      for (BlockInfo b: allBlocks) {
        short replication = b.getReplication();
        long blockSize = b.isComplete() ? b.getNumBytes() : file
            .getPreferredBlockSize();

        oldCounts.addStorageSpace(blockSize * replication);

        if (bsp != null) {
          List<StorageType> oldTypeChosen = bsp.chooseStorageTypes(replication);
          for (StorageType t : oldTypeChosen) {
            if (t.supportTypeQuota()) {
              oldCounts.addTypeSpace(t, blockSize);
            }
          }
        }
      }

      AclFeature aclFeature = removed.getSnapshotINode().getAclFeature();
      if (aclFeature != null) {
        AclStorage.removeAclFeature(aclFeature);
      }
    } else {
      oldCounts = file.storagespaceConsumed(null);
    }

    getDiffs().combineAndCollectSnapshotBlocks(reclaimContext, file, removed);
    if (file.getBlocks() != null) {
      short replInDiff = getMaxBlockRepInDiffs(removed);
      short repl = (short) Math.max(file.getPreferredBlockReplication(),
                                    replInDiff);
      for (BlockInfo b : file.getBlocks()) {
        if (repl != b.getReplication()) {
          reclaimContext.collectedBlocks().addUpdateReplicationFactor(b, repl);
        }
      }
    }
    QuotaCounts current = file.storagespaceConsumed(bsp);
    reclaimContext.quotaDelta().add(oldCounts.subtract(current));
  }
@Test
  public void testUpdateQuotaDistinctBlocks() {
    BlockStoragePolicySuite bsps = mock(BlockStoragePolicySuite.class);
    BlockStoragePolicy bsp = mock(BlockStoragePolicy.class);
    BlockInfo[] blocks = new BlockInfo[] {
        new BlockInfoContiguous(new Block(1, BLOCK_SIZE, 1), REPL_3) };

    INodeFile file = mock(INodeFile.class);
    when(file.getBlocks()).thenReturn(blocks);
    when(file.getStoragePolicyID()).thenReturn((byte) 1);
    when(file.getPreferredBlockReplication()).thenReturn((short) 3);

    when(bsps.getPolicy(anyByte())).thenReturn(bsp);
    INode.BlocksMapUpdateInfo collectedBlocks =
        mock(INode.BlocksMapUpdateInfo.class);
    ArrayList<INode> removedINodes = new ArrayList<>();
    INode.ReclaimContext ctx =
        new INode.ReclaimContext(bsps, collectedBlocks, removedINodes, null);
    QuotaCounts counts = ctx.quotaDelta().getCountsCopy();
    INodeFile snapshotINode = mock(INodeFile.class);

    // add same blocks in file diff
    FileDiff diff1 = new FileDiff(0, snapshotINode, null, 0);
    FileDiff diff = Mockito.spy(diff1);
    Mockito.doReturn(blocks).when(diff).getBlocks();

    // removed file diff
    FileDiff removed = new FileDiff(0, snapshotINode, null, 0);

    // remaining file diffs
    FileDiffList diffs = new FileDiffList();
    diffs.addFirst(diff);
    FileWithSnapshotFeature sf = new FileWithSnapshotFeature(diffs);

    // update quota and collect same blocks in file and file diff
    when(file.getFileWithSnapshotFeature()).thenReturn(sf);
    sf.updateQuotaAndCollectBlocks(ctx, file, removed);
    counts = ctx.quotaDelta().getCountsCopy();
    assertEquals(0, counts.getStorageSpace());

    // different blocks in file and file's diff and in removed diff
    BlockInfo[] blocks1 = new BlockInfo[] {
        new BlockInfoContiguous(new Block(2, BLOCK_SIZE, 1), REPL_3) };
    Mockito.doReturn(blocks1).when(diff).getBlocks();
    // remaining file diffs
    FileDiffList diffs1 = new FileDiffList();
    diffs1.addFirst(diff);
    FileWithSnapshotFeature sf1 = new FileWithSnapshotFeature(diffs1);
    when(file.getFileWithSnapshotFeature()).thenReturn(sf1);
    BlockInfo[] removedBlocks = new BlockInfo[] {
        new BlockInfoContiguous(new Block(3, BLOCK_SIZE, 1), REPL_3) };
    FileDiff removed1 = new FileDiff(0, snapshotINode, null, 1024);
    removed1.setBlocks(removedBlocks);
    INode.ReclaimContext ctx1 =
        new INode.ReclaimContext(bsps, collectedBlocks, removedINodes, null);
    sf1.updateQuotaAndCollectBlocks(ctx1, file, removed1);
    counts = ctx1.quotaDelta().getCountsCopy();
    assertEquals(3072, counts.getStorageSpace());

    // same blocks in file and removed diff
    removed1 = new FileDiff(0, snapshotINode, null, 1024);
    removed1.setBlocks(blocks);
    INode.ReclaimContext ctx2 =
        new INode.ReclaimContext(bsps, collectedBlocks, removedINodes, null);
    sf1.updateQuotaAndCollectBlocks(ctx2, file, removed1);
    counts = ctx2.quotaDelta().getCountsCopy();
    assertEquals(0, counts.getStorageSpace());
  }