ZOOKEEPER-3595: restore the handling of the fsync parameter

Author: Jingguo Yao <yaojingguo@gmail.com>

Reviewers: enixon@apache.org, eolivelli@apache.org, maoling199210191@sina.com, fangmin@apache.org

Closes #1146 from yaojingguo/ZOOKEEPER-3595

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileSnap.java

@@ -231,7 +231,7 @@ public class FileSnap implements SnapShot {
         File snapShot,
         boolean fsync) throws IOException {
         if (!close) {
-            try (CheckedOutputStream snapOS = SnapStream.getOutputStream(snapShot)) {
+            try (CheckedOutputStream snapOS = SnapStream.getOutputStream(snapShot, fsync)) {
                 OutputArchive oa = BinaryOutputArchive.getArchive(snapOS);
                 FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
                 serialize(dt, sessions, oa, header);

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/SnapStream.java

@@ -37,6 +37,7 @@ import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
 import org.apache.jute.InputArchive;
 import org.apache.jute.OutputArchive;
+import org.apache.zookeeper.common.AtomicFileOutputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.xerial.snappy.SnappyCodec;
@@ -119,11 +120,12 @@ public class SnapStream {
      * Return the OutputStream based on predefined stream mode.
      *
      * @param file the file the OutputStream writes to
+     * @param fsync sync the file immediately after write
      * @return the specific OutputStream
      * @throws IOException
      */
-    public static CheckedOutputStream getOutputStream(File file) throws IOException {
-        FileOutputStream fos = new FileOutputStream(file);
+    public static CheckedOutputStream getOutputStream(File file, boolean fsync) throws IOException {
+        OutputStream fos = fsync ? new AtomicFileOutputStream(file) : new FileOutputStream(file);
         OutputStream os;
         switch (streamMode) {
         case GZIP:

zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/SnapStreamTest.java

@@ -79,12 +79,17 @@ public class SnapStreamTest {
     }
 
     private void testSerializeDeserialize(StreamMode mode, String fileSuffix) throws IOException {
+        testSerializeDeserialize(mode, fileSuffix, false);
+        testSerializeDeserialize(mode, fileSuffix, true);
+    }
+
+    private void testSerializeDeserialize(StreamMode mode, String fileSuffix, boolean fsync) throws IOException {
         SnapStream.setStreamMode(mode);
 
         // serialize with gzip stream
         File tmpDir = createTmpDir();
         File file = new File(tmpDir, "snapshot.180000e3a2" + fileSuffix);
-        CheckedOutputStream os = SnapStream.getOutputStream(file);
+        CheckedOutputStream os = SnapStream.getOutputStream(file, fsync);
         OutputArchive oa = BinaryOutputArchive.getArchive(os);
         FileHeader header = new FileHeader(FileSnap.SNAP_MAGIC, 2, 1);
         header.serialize(oa, "fileheader");
@@ -103,20 +108,25 @@ public class SnapStreamTest {
         SnapStream.checkSealIntegrity(is, ia);
     }
 
-    private void checkInvalidSnapshot(String filename) throws IOException {
+    private void checkInvalidSnapshot(String filename, boolean fsync) throws IOException {
         // set the output stream mode to CHECKED
         SnapStream.setStreamMode(StreamMode.CHECKED);
 
         // serialize to CHECKED file without magic header
         File tmpDir = createTmpDir();
         File file = new File(tmpDir, filename);
-        OutputStream os = SnapStream.getOutputStream(file);
+        OutputStream os = SnapStream.getOutputStream(file, fsync);
         os.write(1);
         os.flush();
         os.close();
         assertFalse(SnapStream.isValidSnapshot(file));
     }
 
+    private void checkInvalidSnapshot(String filename) throws IOException {
+        checkInvalidSnapshot(filename, false);
+        checkInvalidSnapshot(filename, true);
+    }
+
     @Test
     public void testInvalidSnapshot() throws IOException {
         assertFalse(SnapStream.isValidSnapshot(null));