Github user nkalmar commented on a diff in the pull request:
https://github.com/apache/zookeeper/pull/690#discussion_r232325878
--- Diff:
zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/SnapStream.java
---
@@ -0,0 +1,336 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.persistence;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.zip.Adler32;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.jute.InputArchive;
+import org.apache.jute.OutputArchive;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xerial.snappy.SnappyCodec;
+import org.xerial.snappy.SnappyInputStream;
+import org.xerial.snappy.SnappyOutputStream;
+
+/**
+ * Represent the Stream used in serialize and deserialize the Snapshot.
+ */
+public class SnapStream {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SnapStream.class);
+
+ public static final String ZOOKEEPER_SHAPSHOT_STREAM_MODE =
+ "zookeeper.snapshot.compression.method";
+
+ private static StreamMode streamMode =
+ StreamMode.fromString(
+ System.getProperty(ZOOKEEPER_SHAPSHOT_STREAM_MODE,
+ StreamMode.DEFAULT_MODE.getName()));
+
+ static {
+ LOG.info(ZOOKEEPER_SHAPSHOT_STREAM_MODE + "=" + streamMode);
+ }
+
+ public static enum StreamMode {
+ GZIP("gz"),
+ SNAPPY("snappy"),
+ CHECKED("");
+
+ public static final StreamMode DEFAULT_MODE = CHECKED;
+
+ private String name;
+
+ StreamMode(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getFileExtension() {
+ return name.isEmpty() ? "" : "." + name;
+ }
+
+ public static StreamMode fromString(String name) {
+ for (StreamMode c : values()) {
+ if (c.getName().compareToIgnoreCase(name) == 0) {
+ return c;
+ }
+ }
+ return DEFAULT_MODE;
+ }
+ }
+
+ /**
+ * Return the CheckedInputStream based on the extension of the
fileName.
+ *
+ * @param fileName the file the InputStream read from
+ * @return the specific InputStream
+ * @throws IOException
+ */
+ public static CheckedInputStream getInputStream(File file) throws
IOException {
+ FileInputStream fis = new FileInputStream(file);
+ InputStream is;
+ switch (getStreamMode(file.getName())) {
+ case GZIP:
+ is = new GZIPInputStream(fis);
+ break;
+ case SNAPPY:
+ is = new SnappyInputStream(fis);
+ break;
+ case CHECKED:
+ default:
+ is = new BufferedInputStream(fis);
+ }
+ return new CheckedInputStream(is, new Adler32());
+ }
+
+ /**
+ * Return the OutputStream based on predefined stream mode.
+ *
+ * @param fileName the file the OutputStream writes to
+ * @return the specific OutputStream
+ * @throws IOException
+ */
+ public static CheckedOutputStream getOutputStream(File file) throws
IOException {
+ FileOutputStream fos = new FileOutputStream(file);
+ OutputStream os;
+ switch (streamMode) {
+ case GZIP:
+ os = new GZIPOutputStream(fos);
+ break;
+ case SNAPPY:
+ os = new SnappyOutputStream(fos);
+ break;
+ case CHECKED:
+ default:
+ os = new BufferedOutputStream(fos);
+ }
+ return new CheckedOutputStream(os, new Adler32());
+ }
+
+ /**
+ * Write specific seal to the OutputArchive and close the OutputStream.
+ * Currently, only CheckedOutputStream will write it's checkSum to the
+ * end of the stream.
+ *
+ */
+ public static void sealStream(CheckedOutputStream os, OutputArchive oa)
+ throws IOException {
+ long val = os.getChecksum().getValue();
+ oa.writeLong(val, "val");
+ oa.writeString("/", "path");
+ }
+
+ /**
+ * Verify the integrity of the seal, only CheckedInputStream will
verify
+ * the checkSum of the content.
+ *
+ */
+ static void checkSealIntegrity(CheckedInputStream is, InputArchive ia)
+ throws IOException {
+ long checkSum = is.getChecksum().getValue();
+ long val = ia.readLong("val");
+ if (val != checkSum) {
+ throw new IOException("CRC corruption");
+ }
+ }
+
+ /**
+ * Verifies that the file is a valid snapshot. Snapshot may be invalid
if
+ * it's incomplete as in a situation when the server dies while in the
+ * process of storing a snapshot. Any files that are improperly
formated
+ * or corrupted are invalid. Any file that is not a snapshot is also an
+ * invalid snapshot.
+ *
+ * @param file file to verify
+ * @return true if the snapshot is valid
+ * @throws IOException
+ */
+ public static boolean isValidSnapshot(File file) throws IOException {
+ if (file == null || Util.getZxidFromName(file.getName(),
FileSnap.SNAPSHOT_FILE_PREFIX) == -1) {
+ return false;
+ }
+
+ String fileName = file.getName();
+ if (Util.getZxidFromName(fileName, "snapshot") == -1) {
+ return false;
+ }
+
+ boolean isValid = false;
+ switch (getStreamMode(fileName)) {
+ case GZIP:
+ isValid = isValidGZipStream(file);
+ break;
+ case SNAPPY:
+ isValid = isValidSnappyStream(file);
+ break;
+ case CHECKED:
+ default:
+ isValid = isValidCheckedStream(file);
+ }
+ return isValid;
+ }
+
+ public static void setStreamMode(StreamMode mode) {
+ streamMode = mode;
+ }
+
+ public static StreamMode getStreamMode() {
+ return streamMode;
+ }
+
+ /**
+ * Detect the stream mode from file name extension
+ *
+ * @param fileName
+ * @return
+ */
+ public static StreamMode getStreamMode(String fileName) {
+ String[] splitSnapName = fileName.split("\\.");
+
+ // Use file extension to detect format
+ if (splitSnapName.length > 1) {
+ String mode = splitSnapName[splitSnapName.length - 1];
+ return StreamMode.fromString(mode);
+ }
+
+ return StreamMode.CHECKED;
+ }
+
+ /**
+ * Certify the GZip stream integrity by checking the header
+ * for the GZip magic string
+ *
+ * @param f file to verify
+ * @return true if it has the correct GZip magic string
+ * @throws IOException
+ */
+ private static boolean isValidGZipStream(File f) throws IOException {
+ FileInputStream fis = null;
+ byte[] byteArray = new byte[2];
+
+ try {
+ fis = new FileInputStream(f);
+ fis.read(byteArray, 0, 2);
+ ByteBuffer bb = ByteBuffer.wrap(byteArray);
+ byte[] magicHeader = new byte[2];
+ bb.get(magicHeader, 0, 2);
+ int magic = magicHeader[0] & 0xff | ((magicHeader[1] << 8) &
0xff00);
+ return magic == GZIPInputStream.GZIP_MAGIC;
+ } catch (FileNotFoundException e) {
+ LOG.error("Unable to open file " + f.getName() + " : ", e);
+ return false;
+ } finally {
+ if (fis != null) {
+ fis.close();
+ }
+ }
+ }
+
+ /**
+ * Certify the Snappy stream integrity by checking the header
+ * for the Snappy magic string
+ *
+ * @param f file to verify
+ * @return true if it has the correct Snappy magic string
+ * @throws IOException
+ */
+ private static boolean isValidSnappyStream(File f) throws IOException {
+ FileInputStream fis = null;
+ byte[] byteArray = new byte[SnappyCodec.MAGIC_LEN];
+ try {
+ fis = new FileInputStream(f);
+ fis.read(byteArray, 0, SnappyCodec.MAGIC_LEN);
--- End diff --
Again, findBugs doesn't like it. See above.
---
