HDDS-12397. Persist putBlock for closed container.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -1099,11 +1099,16 @@ public void writeChunkForClosedContainer(ChunkInfo chunkInfo, BlockID blockID,
 
   /**
    * Handle Put Block operation for closed container. Calls BlockManager to process the request.
-   *
+   * This is primarily used by container reconciliation process to persist the block data for closed container.
+   * @param kvContainer - Container for which block data need to be persisted.
+   * @param blockData - Block Data to be persisted (BlockData should have the chunks).
+   * @param blockCommitSequenceId - Block Commit Sequence ID for the block.
+   * @param overwriteBscId - To overwrite bcsId in the block data. In case of chunk failure during reconciliation,
+   *                      we do not want to overwrite the bcsId as this block is incomplete in its current state.
    */
-  public void putBlockForClosedContainer(List<ContainerProtos.ChunkInfo> chunkInfos, KeyValueContainer kvContainer,
-                                          BlockData blockData, long blockCommitSequenceId)
-      throws IOException {
+  public void putBlockForClosedContainer(KeyValueContainer kvContainer, BlockData blockData,
+                                         long blockCommitSequenceId, boolean overwriteBscId)
+          throws IOException {
     Preconditions.checkNotNull(kvContainer);
     Preconditions.checkNotNull(blockData);
     long startTime = Time.monotonicNowNanos();
@@ -1112,11 +1117,12 @@ public void putBlockForClosedContainer(List<ContainerProtos.ChunkInfo> chunkInfo
       throw new IOException("Container #" + kvContainer.getContainerData().getContainerID() +
           " is not in closed state, Container state is " + kvContainer.getContainerState());
     }
-    blockData.setChunks(chunkInfos);
     // To be set from the Replica's BCSId
-    blockData.setBlockCommitSequenceId(blockCommitSequenceId);
+    if (overwriteBscId) {
+      blockData.setBlockCommitSequenceId(blockCommitSequenceId);
+    }
 
-    blockManager.putBlock(kvContainer, blockData, false);
+    blockManager.putBlockForClosedContainer(kvContainer, blockData, overwriteBscId);
     ContainerProtos.BlockData blockDataProto = blockData.getProtoBufMessage();
     final long numBytes = blockDataProto.getSerializedSize();
     // Increment write stats for PutBlock after write.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java

@@ -98,6 +98,81 @@ public long putBlock(Container container, BlockData data,
         data, endOfBlock);
   }
 
+  @Override
+  public long putBlockForClosedContainer(Container container, BlockData data, boolean overwriteBcsId)
+          throws IOException {
+    Preconditions.checkNotNull(data, "BlockData cannot be null for put " +
+        "operation.");
+    Preconditions.checkState(data.getContainerID() >= 0, "Container Id " +
+        "cannot be negative");
+
+    KeyValueContainerData containerData = (KeyValueContainerData) container.getContainerData();
+
+    // We are not locking the key manager since RocksDB serializes all actions
+    // against a single DB. We rely on DB level locking to avoid conflicts.
+    try (DBHandle db = BlockUtils.getDB(containerData, config)) {
+      // This is a post condition that acts as a hint to the user.
+      // Should never fail.
+      Preconditions.checkNotNull(db, DB_NULL_ERR_MSG);
+
+      long bcsId = data.getBlockCommitSequenceId();
+      long containerBCSId = containerData.getBlockCommitSequenceId();
+
+      // Check if the block is already present in the DB of the container to determine whether
+      // the blockCount is already incremented for this block in the DB or not.
+      long localID = data.getLocalID();
+      boolean incrBlockCount = false;
+
+      // update the blockData as well as BlockCommitSequenceId here
+      try (BatchOperation batch = db.getStore().getBatchHandler()
+          .initBatchOperation()) {
+        // If block exists in cache, blockCount should not be incremented.
+        if (db.getStore().getBlockDataTable().get(containerData.getBlockKey(localID)) == null) {
+          // Block does not exist in DB => blockCount needs to be
+          // incremented when the block is added into DB.
+          incrBlockCount = true;
+        }
+
+        db.getStore().getBlockDataTable().putWithBatch(batch, containerData.getBlockKey(localID), data);
+        if (overwriteBcsId && bcsId > containerBCSId) {
+          db.getStore().getMetadataTable().putWithBatch(batch, containerData.getBcsIdKey(), bcsId);
+        }
+
+        // Set Bytes used, this bytes used will be updated for every write and
+        // only get committed for every put block. In this way, when datanode
+        // is up, for computation of disk space by container only committed
+        // block length is used, And also on restart the blocks committed to DB
+        // is only used to compute the bytes used. This is done to keep the
+        // current behavior and avoid DB write during write chunk operation.
+        db.getStore().getMetadataTable().putWithBatch(batch, containerData.getBytesUsedKey(),
+            containerData.getBytesUsed());
+
+        // Set Block Count for a container.
+        if (incrBlockCount) {
+          db.getStore().getMetadataTable().putWithBatch(batch, containerData.getBlockCountKey(),
+              containerData.getBlockCount() + 1);
+        }
+
+        db.getStore().getBatchHandler().commitBatchOperation(batch);
+      }
+
+      if (overwriteBcsId && bcsId > containerBCSId) {
+        container.updateBlockCommitSequenceId(bcsId);
+      }
+
+      // Increment block count in-memory after the DB update.
+      if (incrBlockCount) {
+        containerData.incrBlockCount();
+      }
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Block {} successfully persisted for closed container {} with bcsId {} chunk size {}",
+            data.getBlockID(), containerData.getContainerID(), bcsId, data.getChunks().size());
+      }
+      return data.getSize();
+    }
+  }
+
   public long persistPutBlock(KeyValueContainer container,
       BlockData data, boolean endOfBlock)
       throws IOException {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/FilePerBlockStrategy.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.keyvalue.impl;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CHUNK_FILE_INCONSISTENCY;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNSUPPORTED_REQUEST;
 import static org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion.FILE_PER_BLOCK;
 import static org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext.WriteChunkStage.COMMIT_DATA;
@@ -174,8 +175,25 @@ public void writeChunk(Container container, BlockID blockID, ChunkInfo info,
       ChunkUtils.validateChunkSize(channel, info, chunkFile.getName());
     }
 
-    ChunkUtils
-        .writeData(channel, chunkFile.getName(), data, offset, len, volume);
+    long fileLengthBeforeWrite;
+    try {
+      fileLengthBeforeWrite = channel.size();
+    } catch (IOException e) {
+      throw new StorageContainerException("IO error encountered while " +
+          "getting the file size for " + chunkFile.getName(), CHUNK_FILE_INCONSISTENCY);
+    }
+
+    ChunkUtils.writeData(channel, chunkFile.getName(), data, offset, len, volume);
+
+    // When overwriting, update the bytes used if the new length is greater than the old length
+    // This is to ensure that the bytes used is updated correctly when overwriting a smaller chunk
+    // with a larger chunk.
+    if (overwrite) {
+      long fileLengthAfterWrite = offset + len;
+      if (fileLengthAfterWrite > fileLengthBeforeWrite) {
+        containerData.incrBytesUsed(fileLengthAfterWrite - fileLengthBeforeWrite);
+      }
+    }
 
     containerData.updateWriteStats(len, overwrite);
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/interfaces/BlockManager.java

@@ -49,6 +49,17 @@ public interface BlockManager {
   long putBlock(Container container, BlockData data, boolean endOfBlock)
       throws IOException;
 
+  /**
+   * Puts or overwrites a block to a closed container.
+   *
+   * @param container - Container for which block need to be added.
+   * @param data - Block Data.
+   * @param overwriteBcsId - To overwrite bcsId in the block data.
+   * @return length of the Block.
+   */
+  long putBlockForClosedContainer(Container container, BlockData data, boolean overwriteBcsId)
+          throws IOException;
+
   /**
    * Gets an existing block.
    *

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.keyvalue.impl;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED;
 import static org.apache.hadoop.ozone.OzoneConsts.INCREMENTAL_CHUNK_LIST;
 import static org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil.isSameSchemaVersion;
 import static org.apache.hadoop.ozone.container.keyvalue.impl.BlockManagerImpl.FULL_CHUNK;
@@ -36,10 +37,12 @@
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.helpers.BlockData;
 import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
 import org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion;
+import org.apache.hadoop.ozone.container.common.interfaces.DBHandle;
 import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
@@ -50,6 +53,7 @@
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Assumptions;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -81,10 +85,10 @@ private void initTest(ContainerTestVersionInfo versionInfo)
     this.schemaVersion = versionInfo.getSchemaVersion();
     this.config = new OzoneConfiguration();
     ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, config);
-    initilaze();
+    initialize();
   }
 
-  private void initilaze() throws Exception {
+  private void initialize() throws Exception {
     UUID datanodeId = UUID.randomUUID();
     HddsVolume hddsVolume = new HddsVolume.Builder(folder.toString())
         .conf(config)
@@ -196,6 +200,65 @@ public void testPutAndGetBlock(ContainerTestVersionInfo versionInfo)
 
   }
 
+  @ContainerTestVersionInfo.ContainerTest
+  public void testPutBlockForClosed(ContainerTestVersionInfo versionInfo)
+      throws Exception {
+    initTest(versionInfo);
+    KeyValueContainerData containerData = keyValueContainer.getContainerData();
+    assertEquals(0, containerData.getBlockCount());
+    keyValueContainer.close();
+    assertEquals(CLOSED, keyValueContainer.getContainerState());
+    // 1. Put Block with bcsId = 2, Overwrite = true
+    blockManager.putBlockForClosedContainer(keyValueContainer, blockData1, true);
+
+    try (DBHandle db = BlockUtils.getDB(containerData, config)) {
+      BlockData fromGetBlockData;
+      //Check Container's bcsId
+      fromGetBlockData = blockManager.getBlock(keyValueContainer, blockData1.getBlockID());
+      assertEquals(1, containerData.getBlockCount());
+      assertEquals(1, containerData.getBlockCommitSequenceId());
+      assertEquals(1, fromGetBlockData.getBlockCommitSequenceId());
+      assertEquals(1, db.getStore().getMetadataTable().get(containerData.getBcsIdKey()));
+      assertEquals(1, db.getStore().getMetadataTable().get(containerData.getBlockCountKey()));
+
+      // 2. Put Block with bcsId = 2, Overwrite = false
+      BlockData blockData2 = createBlockData(1L, 3L, 1, 0, 2048, 2);
+      blockManager.putBlockForClosedContainer(keyValueContainer, blockData2, false);
+
+      // The block should be written, but we won't be able to read it, As BcsId < container's BcsId
+      // fails during block read.
+      Assertions.assertThrows(StorageContainerException.class, () -> blockManager
+          .getBlock(keyValueContainer, blockData2.getBlockID()));
+      assertEquals(2, containerData.getBlockCount());
+      // BcsId should still be 1, as the BcsId is not overwritten
+      assertEquals(1, containerData.getBlockCommitSequenceId());
+      assertEquals(2, db.getStore().getMetadataTable().get(containerData.getBlockCountKey()));
+      assertEquals(1, db.getStore().getMetadataTable().get(containerData.getBcsIdKey()));
+
+      // 3. Put Block with bcsId = 2, Overwrite = true
+      // This should succeed as we are overwriting the BcsId, The container BcsId should be updated to 3
+      // The block count should not change.
+      blockManager.putBlockForClosedContainer(keyValueContainer, blockData2, true);
+      fromGetBlockData = blockManager.getBlock(keyValueContainer, blockData2.getBlockID());
+      assertEquals(2, containerData.getBlockCount());
+      assertEquals(2, containerData.getBlockCommitSequenceId());
+      assertEquals(2, fromGetBlockData.getBlockCommitSequenceId());
+      assertEquals(2, db.getStore().getMetadataTable().get(containerData.getBlockCountKey()));
+      assertEquals(2, db.getStore().getMetadataTable().get(containerData.getBcsIdKey()));
+
+      // 4. Put Block with bcsId = 1 < container bcsId, Overwrite = true
+      // Container bcsId should not change
+      BlockData blockData3 = createBlockData(1L, 1L, 1, 0, 2048, 1);
+      blockManager.putBlockForClosedContainer(keyValueContainer, blockData3, true);
+      fromGetBlockData = blockManager.getBlock(keyValueContainer, blockData3.getBlockID());
+      assertEquals(3, containerData.getBlockCount());
+      assertEquals(2, containerData.getBlockCommitSequenceId());
+      assertEquals(1, fromGetBlockData.getBlockCommitSequenceId());
+      assertEquals(3, db.getStore().getMetadataTable().get(containerData.getBlockCountKey()));
+      assertEquals(2, db.getStore().getMetadataTable().get(containerData.getBcsIdKey()));
+    }
+  }
+
   @ContainerTestVersionInfo.ContainerTest
   public void testListBlock(ContainerTestVersionInfo versionInfo)
       throws Exception {

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestFilePerBlockStrategy.java

@@ -164,8 +164,8 @@ public void testWriteChunkAndPutBlockFailureForNonClosedContainer(
     ChunkBuffer.wrap(getData());
     Assertions.assertThrows(IOException.class, () -> keyValueHandler.writeChunkForClosedContainer(
         getChunkInfo(), getBlockID(), ChunkBuffer.wrap(getData()), keyValueContainer));
-    Assertions.assertThrows(IOException.class, () -> keyValueHandler.putBlockForClosedContainer(
-        null, keyValueContainer, new BlockData(getBlockID()), 0L));
+    Assertions.assertThrows(IOException.class, () -> keyValueHandler.putBlockForClosedContainer(keyValueContainer,
+            new BlockData(getBlockID()), 0L, true));
   }
 
   @Test
@@ -225,37 +225,70 @@ public void testPutBlockForClosedContainer() throws IOException {
     containerSet.addContainer(kvContainer);
     KeyValueHandler keyValueHandler = createKeyValueHandler(containerSet);
     List<ContainerProtos.ChunkInfo> chunkInfoList = new ArrayList<>();
-    chunkInfoList.add(getChunkInfo().getProtoBufMessage());
+    ChunkInfo info = new ChunkInfo(String.format("%d.data.%d", getBlockID().getLocalID(), 0), 0L, 20L);
+
+    chunkInfoList.add(info.getProtoBufMessage());
     BlockData putBlockData = new BlockData(getBlockID());
-    keyValueHandler.putBlockForClosedContainer(chunkInfoList, kvContainer, putBlockData, 1L);
-    Assertions.assertEquals(containerData.getBlockCommitSequenceId(), 1L);
-    Assertions.assertEquals(containerData.getBlockCount(), 1L);
+    putBlockData.setChunks(chunkInfoList);
+
+    ChunkBuffer chunkData = ContainerTestHelper.getData(20);
+    keyValueHandler.writeChunkForClosedContainer(info, getBlockID(), chunkData, kvContainer);
+    keyValueHandler.putBlockForClosedContainer(kvContainer, putBlockData, 1L, true);
+    assertEquals(1L, containerData.getBlockCommitSequenceId());
+    assertEquals(1L, containerData.getBlockCount());
 
     try (DBHandle dbHandle = BlockUtils.getDB(containerData, new OzoneConfiguration())) {
       long localID = putBlockData.getLocalID();
       BlockData getBlockData = dbHandle.getStore().getBlockDataTable()
           .get(containerData.getBlockKey(localID));
       Assertions.assertTrue(blockDataEquals(putBlockData, getBlockData));
+      // Overwriting the same
+      assertEquals(20L, containerData.getBytesUsed());
+      assertEquals(20L, dbHandle.getStore().getMetadataTable().get(containerData.getBytesUsedKey()));
     }
 
     // Add another chunk and check the put block data
-    ChunkInfo newChunkInfo = new ChunkInfo(String.format("%d.data.%d", getBlockID()
-        .getLocalID(), 1L), 0, 20L);
+    ChunkInfo newChunkInfo = new ChunkInfo(String.format("%d.data.%d", getBlockID().getLocalID(), 1L), 20L, 20L);
+    chunkInfoList.add(newChunkInfo.getProtoBufMessage());
+    putBlockData.setChunks(chunkInfoList);
+
+    chunkData = ContainerTestHelper.getData(20);
+    keyValueHandler.writeChunkForClosedContainer(newChunkInfo, getBlockID(), chunkData, kvContainer);
+    keyValueHandler.putBlockForClosedContainer(kvContainer, putBlockData, 2L, true);
+    assertEquals(2L, containerData.getBlockCommitSequenceId());
+    assertEquals(1L, containerData.getBlockCount());
+
+    try (DBHandle dbHandle = BlockUtils.getDB(containerData, new OzoneConfiguration())) {
+      long localID = putBlockData.getLocalID();
+      BlockData getBlockData = dbHandle.getStore().getBlockDataTable()
+          .get(containerData.getBlockKey(localID));
+      Assertions.assertTrue(blockDataEquals(putBlockData, getBlockData));
+      assertEquals(40L, containerData.getBytesUsed());
+      assertEquals(40L, dbHandle.getStore().getMetadataTable().get(containerData.getBytesUsedKey()));
+    }
+
+    newChunkInfo = new ChunkInfo(String.format("%d.data.%d", getBlockID().getLocalID(), 1L), 20L, 30L);
+    chunkInfoList.remove(chunkInfoList.size() - 1);
     chunkInfoList.add(newChunkInfo.getProtoBufMessage());
-    keyValueHandler.putBlockForClosedContainer(chunkInfoList, kvContainer, putBlockData, 2L);
-    Assertions.assertEquals(containerData.getBlockCommitSequenceId(), 2L);
-    Assertions.assertEquals(containerData.getBlockCount(), 1L);
+    putBlockData.setChunks(chunkInfoList);
+
+    chunkData = ContainerTestHelper.getData(30);
+    keyValueHandler.writeChunkForClosedContainer(newChunkInfo, getBlockID(), chunkData, kvContainer);
+    keyValueHandler.putBlockForClosedContainer(kvContainer, putBlockData, 2L, true);
+    assertEquals(2L, containerData.getBlockCommitSequenceId());
+    assertEquals(1L, containerData.getBlockCount());
 
     try (DBHandle dbHandle = BlockUtils.getDB(containerData, new OzoneConfiguration())) {
       long localID = putBlockData.getLocalID();
       BlockData getBlockData = dbHandle.getStore().getBlockDataTable()
           .get(containerData.getBlockKey(localID));
       Assertions.assertTrue(blockDataEquals(putBlockData, getBlockData));
+      assertEquals(50L, containerData.getBytesUsed());
+      assertEquals(50L, dbHandle.getStore().getMetadataTable().get(containerData.getBytesUsedKey()));
     }
 
-    // Put block on bcsId <= containerBcsId should be a no-op
-    keyValueHandler.putBlockForClosedContainer(chunkInfoList, kvContainer, putBlockData, 2L);
-    Assertions.assertEquals(containerData.getBlockCommitSequenceId(), 2L);
+    keyValueHandler.putBlockForClosedContainer(kvContainer, putBlockData, 2L, true);
+    assertEquals(2L, containerData.getBlockCommitSequenceId());
   }
 
   private boolean blockDataEquals(BlockData putBlockData, BlockData getBlockData) {