HDDS-14246. Change fsync boundary for FilePerBlockStrategy to block level

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

@@ -85,8 +85,9 @@
     <name>hdds.container.chunk.write.sync</name>
     <value>false</value>
     <tag>OZONE, CONTAINER, MANAGEMENT</tag>
-    <description>Determines whether the chunk writes in the container happen as
-      sync I/0 or buffered I/O operation.
+    <description>For the deprecated FilePerChunkStrategy, this determines whether the chunk writes
+      in the container happen as sync I/0 or buffered I/O operation. For FilePerBlockStrategy, this
+      the sync I/O operation only happens before block file is closed.
     </description>
   </property>
   <property>

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

@@ -671,6 +671,11 @@ ContainerCommandResponseProto handlePutBlock(
           chunkManager.finishWriteChunks(kvContainer, blockData);
         }
         endOfBlock = true;
+      } else {
+        // If sync I/O is enabled, we should sync the changes made by the previous WriteChunk(s)
+        // to ensure that once PutBlock is complete, the data has been synced to the underlying
+        // persistent storage. Note that finishWriteChunks will also trigger sync I/O if enabled.
+        chunkManager.syncChunks(kvContainer, blockData.getBlockID());
       }
 
       // Note: checksum held inside blockData. But no extra checksum validation here with handlePutBlock.

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

@@ -140,6 +140,11 @@ public void deleteChunks(Container container, BlockData blockData)
     selectHandler(container).deleteChunks(container, blockData);
   }
 
+  @Override
+  public void syncChunks(Container container, BlockID blockID) throws IOException {
+    selectHandler(container).syncChunks(container, blockID);
+  }
+
   @Override
   public void shutdown() {
     handlers.values().forEach(ChunkManager::shutdown);

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

@@ -270,6 +270,17 @@ public void finalizeWriteChunk(KeyValueContainer container,
     }
   }
 
+  @Override
+  public void syncChunks(Container container, BlockID blockID) throws IOException {
+    final File chunkFile = getChunkFile(container, blockID);
+    try {
+      files.sync(chunkFile);
+    } catch (IOException e) {
+      onFailure(container.getContainerData().getVolume());
+      throw e;
+    }
+  }
+
   private void deleteChunk(Container container, BlockID blockID,
       ChunkInfo info, boolean verifyLength)
       throws StorageContainerException {
@@ -367,15 +378,28 @@ private static void close(String filename, OpenFile openFile) {
         }
       }
     }
+
+    public void sync(File file) throws IOException {
+      if (file != null) {
+        OpenFile openFile = files.getIfPresent(file.getPath());
+        if (openFile != null) {
+          openFile.syncIfNeeded();
+        }
+      }
+    }
   }
 
   private static final class OpenFile {
 
     private final RandomAccessFile file;
+    private final boolean sync;
 
     private OpenFile(File file, boolean sync) throws FileNotFoundException {
-      String mode = sync ? "rws" : "rw";
-      this.file = new RandomAccessFile(file, mode);
+      // We should only trigger fsync before the block file is closed (before PutBlock)
+      // instead of every write (by using "rws") since block is only visible to user after PutBlock
+      // and doing sync I/O for each write will cause unnecessary I/O overhead.
+      this.file = new RandomAccessFile(file, "rw");
+      this.sync = sync;
       if (LOG.isDebugEnabled()) {
         LOG.debug("Opened file {}", file);
       }
@@ -385,8 +409,16 @@ public FileChannel getChannel() {
       return file.getChannel();
     }
 
+    private void syncIfNeeded() throws IOException {
+      if (sync) {
+        // ensure data and metadata is persisted
+        getChannel().force(true);
+      }
+    }
+
     public void close() {
       try {
+        syncIfNeeded();
         file.close();
       } catch (IOException e) {
         throw new UncheckedIOException(e);

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

@@ -108,6 +108,10 @@ default void finalizeWriteChunk(KeyValueContainer container,
     // no-op
   }
 
+  default void syncChunks(Container container, BlockID blockID) throws IOException {
+    // no-op
+  }
+
   default String streamInit(Container container, BlockID blockID)
       throws StorageContainerException {
     return null;