ZOOKEEPER-121. SyncRequestProcessor is not closing log stream during shutdown (mahadev)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@922442 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -273,6 +273,9 @@ BUGFIXES:
   ZOOKEEPER-604. zk needs to prevent export of any symbol not listed in their
   api (mahadev)
 
+  ZOOKEEPER-121. SyncRequestProcessor is not closing log stream during
+  shutdown (mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-473. cleanup junit tests to eliminate false positives due to
   "socket reuse" and failure to close client (phunt via mahadev)

src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java

@@ -166,6 +166,11 @@ public class SyncRequestProcessor extends Thread implements RequestProcessor {
     }
 
     public void shutdown() {
+        try{
+            zks.getZKDatabase().close();
+        } catch (IOException ie) {
+            LOG.warn("Error closing logs ", ie);
+        }
         queuedRequests.add(requestOfDeath);
         nextProcessor.shutdown();
     }

src/java/main/org/apache/zookeeper/server/ZKDatabase.java

@@ -450,5 +450,12 @@ public class ZKDatabase {
         this.snapLog.commit();
     }
     
+    /**
+     * close this database. free the resources
+     * @throws IOException
+     */
+    public void close() throws IOException {
+        this.snapLog.close();
+    }
     
 }

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

@@ -49,6 +49,7 @@ import org.apache.zookeeper.server.util.SerializeUtils;
  */
 public class FileSnap implements SnapShot {
     File snapDir;
+    private volatile boolean close = false;
     private static final int VERSION=2;
     private static final long dbId=-1;
     private static final Logger LOG = Logger.getLogger(FileSnap.class);
@@ -217,20 +218,32 @@ public class FileSnap implements SnapShot {
      * @param sessions the sessions to be serialized
      * @param snapShot the file to store snapshot into
      */
-    public void serialize(DataTree dt, Map<Long, Integer> sessions, File snapShot)
+    public synchronized void serialize(DataTree dt, Map<Long, Integer> sessions, File snapShot)
             throws IOException {
-        OutputStream sessOS = new BufferedOutputStream(new FileOutputStream(snapShot));
-        CheckedOutputStream crcOut = new CheckedOutputStream(sessOS, new Adler32());
-        //CheckedOutputStream cout = new CheckedOutputStream()
-        OutputArchive oa = BinaryOutputArchive.getArchive(crcOut);
-        FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
-        serialize(dt,sessions,oa, header);
-        long val = crcOut.getChecksum().getValue();
-        oa.writeLong(val, "val");
-        oa.writeString("/", "path");
-        sessOS.flush();
-        crcOut.close();
-        sessOS.close();
+        if (!close) {
+            OutputStream sessOS = new BufferedOutputStream(new FileOutputStream(snapShot));
+            CheckedOutputStream crcOut = new CheckedOutputStream(sessOS, new Adler32());
+            //CheckedOutputStream cout = new CheckedOutputStream()
+            OutputArchive oa = BinaryOutputArchive.getArchive(crcOut);
+            FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
+            serialize(dt,sessions,oa, header);
+            long val = crcOut.getChecksum().getValue();
+            oa.writeLong(val, "val");
+            oa.writeString("/", "path");
+            sessOS.flush();
+            crcOut.close();
+            sessOS.close();
+        }
+    }
+
+    /**
+     * synchronized close just so that if serialize is in place
+     * the close operation will block and will wait till serialize
+     * is done and will set the close flag
+     */
+    @Override
+    public synchronized void close() throws IOException {
+        close = true;
     }
 
  }

src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java

@@ -164,6 +164,19 @@ public class FileTxnLog implements TxnLog {
         }
     }
 
+    /**
+     * close all the open file handles
+     * @throws IOException
+     */
+    public synchronized void close() throws IOException {
+        if (logStream != null) {
+            logStream.close();
+        }
+        for (FileOutputStream log : streamsToFlush) {
+            log.close();
+        }
+    }
+    
     /**
      * append an entry to the transaction log
      * @param hdr the header of the transaction

src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java

@@ -285,4 +285,12 @@ public class FileTxnSnapLog {
         txnLog.rollLog();
     }
     
+    /**
+     * close the transaction log files
+     * @throws IOException
+     */
+    public void close() throws IOException {
+        txnLog.close();
+        snapLog.close();
+    }
 }

src/java/main/org/apache/zookeeper/server/persistence/SnapShot.java

@@ -57,4 +57,10 @@ public interface SnapShot {
      * @throws IOException
      */
     File findMostRecentSnapshot() throws IOException;
+    
+    /**
+     * free resources from this snapshot immediately
+     * @throws IOException
+     */
+    void close() throws IOException;
 } 

src/java/main/org/apache/zookeeper/server/persistence/TxnLog.java

@@ -83,6 +83,10 @@ public interface TxnLog {
      */
     void commit() throws IOException;
    
+    /** 
+     * close the transactions logs
+     */
+    void close() throws IOException;
     /**
      * an iterating interface for reading 
      * transaction logs. 

src/java/test/org/apache/zookeeper/test/ClientBase.java

@@ -454,9 +454,7 @@ public abstract class ClientBase extends TestCase {
         stopServer();
 
         if (tmpDir != null) {
-            //assertTrue("delete " + tmpDir.toString(), recursiveDelete(tmpDir));
-            // FIXME see ZOOKEEPER-121 replace following line with previous
-            recursiveDelete(tmpDir);
+            assertTrue("delete " + tmpDir.toString(), recursiveDelete(tmpDir));
         }
 
         // This has to be set to null when the same instance of this class is reused between test cases
@@ -475,9 +473,7 @@ public abstract class ClientBase extends TestCase {
         if (d.isDirectory()) {
             File children[] = d.listFiles();
             for (File f : children) {
-                //assertTrue("delete " + f.toString(), recursiveDelete(f));
-                // FIXME see ZOOKEEPER-121 replace following line with previous
-                recursiveDelete(f);
+                assertTrue("delete " + f.toString(), recursiveDelete(f));
             }
         }
         return d.delete();