ZOOKEEPER-4246: Resource leaks in org.apache.zookeeper.server.persist…

…ence.SnapStream#getInputStream and #getOutputStream

Bug report is here: https://issues.apache.org/jira/browse/ZOOKEEPER-4246

This fix is simple: it just closes the possibly-leaked streams and re-throws the exception. We can't use a try-with-resources or a `finally` block here because in the happy case the resulting streams need to be returned open. I checked each of the other constructor calls in these methods, and none of the others can throw an exception as far as I can tell.

Author: Martin Kellogg <kelloggm@cs.washington.edu>

Reviewers: Andor Molnar <anmolnar@apache.org>, Enrico Olivelli <eolivelli@apache.org>, maoling <maoling@apache.org>

Closes apache#1638 from kelloggm/ZOOKEEPER-4246 and squashes the following commits:

fa1c0f0 [Martin Kellogg] surround whole switch with one try block instead of two inside the switch
f023851 [Martin Kellogg] remove all the tabs for real
04c4b2f [Martin Kellogg] fix accidental tab
e896efa [Martin Kellogg] ZOOKEEPER-4246: Resource leaks in org.apache.zookeeper.server.persistence.SnapStream#getInputStream and #getOutputStream

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

@@ -102,18 +102,23 @@ public static StreamMode fromString(String name) {
     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);
+        try {
+            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());
+        } catch (IOException e) {
+            fis.close();
+            throw e;
         }
-        return new CheckedInputStream(is, new Adler32());
     }
 
     /**
@@ -129,9 +134,16 @@ public static CheckedOutputStream getOutputStream(File file, boolean fsync) thro
         OutputStream os;
         switch (streamMode) {
         case GZIP:
-            os = new GZIPOutputStream(fos);
+            try {
+                os = new GZIPOutputStream(fos);
+            } catch (IOException e) {
+                fos.close();
+                throw e;
+            }
             break;
         case SNAPPY:
+            // Unlike SnappyInputStream, the SnappyOutputStream
+            // constructor cannot throw an IOException.
             os = new SnappyOutputStream(fos);
             break;
         case CHECKED: