HDDS-1654. Ensure container state on datanode gets synced to disk whennever state change happens.

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

@@ -80,20 +80,25 @@ private 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);
       // Compute Checksum and update ContainerData
       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

@@ -249,6 +249,9 @@ public long takeSnapshot() throws IOException {
       LOG.info("Taking a snapshot to file {}", 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.warn("Failed to write snapshot file \"" + snapshotFile
             + "\", last applied index=" + ti);