HDDS-1654. Ensure container state on datanode gets synced to disk whennever state change happens. Cotributed by Shashikant Banerjee. (#923)

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java

@@ -80,6 +80,7 @@ public final class ContainerDataYaml {
   public static void createContainerFile(ContainerType containerType,
       ContainerData containerData, File containerFile) throws IOException {
     Writer writer = null;
+    FileOutputStream out = null;
     try {
       // Create Yaml for given container type
       Yaml yaml = getYamlForContainerType(containerType);
@@ -87,13 +88,17 @@ public final class ContainerDataYaml {
       containerData.computeAndSetChecksum(yaml);
 
       // Write the ContainerData with checksum to Yaml file.
-      writer = new OutputStreamWriter(new FileOutputStream(
-          containerFile), "UTF-8");
+      out = new FileOutputStream(
+          containerFile);
+      writer = new OutputStreamWriter(out, "UTF-8");
       yaml.dump(containerData, writer);
 
     } finally {
       try {
         if (writer != null) {
+          writer.flush();
+          // make sure the container metadata is synced to disk.
+          out.getFD().sync();
           writer.close();
         }
       } catch (IOException ex) {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java

@@ -262,6 +262,9 @@ public class ContainerStateMachine extends BaseStateMachine {
       LOG.info("{}: Taking a snapshot at:{} file {}", gid, ti, snapshotFile);
       try (FileOutputStream fos = new FileOutputStream(snapshotFile)) {
         persistContainerSet(fos);
+        fos.flush();
+        // make sure the snapshot file is synced
+        fos.getFD().sync();
       } catch (IOException ioe) {
         LOG.info("{}: Failed to write snapshot at:{} file {}", gid, ti,
             snapshotFile);