ZOOKEEPER-1844. TruncateTest fails on windows (Rakesh R via fpj)

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

CHANGES.txt

@@ -553,6 +553,8 @@ BUGFIXES:
   ZOOKEEPER-1873. Unnecessarily InstanceNotFoundException is coming when
   unregister failed jmxbeans (Rakesh R via michim)
 
+  ZOOKEEPER-1844. TruncateTest fails on windows (Rakesh R via fpj)
+
 IMPROVEMENTS:
 
   ZOOKEEPER-1170. Fix compiler (eclipse) warnings: unused imports,

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

@@ -95,6 +95,20 @@ public class TxnLogProposalIterator implements Iterator<Proposal> {
         throw new UnsupportedOperationException();
     }
 
+    /**
+     * Close the files and release the resources which are used for iterating
+     * transaction records
+     */
+    public void close() {
+        if(itr != null){
+            try {
+                itr.close();
+            } catch (IOException ioe) {
+                LOG.warn("Error closing file iterator", ioe);
+            }
+        }
+    }
+
     private TxnLogProposalIterator() {
     }
 

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

@@ -307,6 +307,7 @@ public class ZKDatabase {
                     && (itr.getHeader().getZxid() > startZxid)) {
                 LOG.warn("Unable to find proposals from txnlog for zxid: "
                         + startZxid);
+                itr.close();
                 return TxnLogProposalIterator.EMPTY_ITERATOR;
             }
 
@@ -315,11 +316,19 @@ public class ZKDatabase {
                 if (txnSize > sizeLimit) {
                     LOG.info("Txnlog size: " + txnSize + " exceeds sizeLimit: "
                             + sizeLimit);
+                    itr.close();
                     return TxnLogProposalIterator.EMPTY_ITERATOR;
                 }
             }
         } catch (IOException e) {
             LOG.error("Unable to read txnlog from disk", e);
+            try {
+                if (itr != null) {
+                    itr.close();
+                }
+            } catch (IOException ioe) {
+                LOG.warn("Error closing file iterator", ioe);
+            }
             return TxnLogProposalIterator.EMPTY_ITERATOR;
         }
         return new TxnLogProposalIterator(itr);

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

@@ -286,9 +286,10 @@ public class FileTxnLog implements TxnLog {
         // if a log file is more recent we must scan it to find
         // the highest zxid
         long zxid = maxLog;
+        TxnIterator itr = null;
         try {
             FileTxnLog txn = new FileTxnLog(logDir);
-            TxnIterator itr = txn.read(maxLog);
+            itr = txn.read(maxLog);
             while (true) {
                 if(!itr.next())
                     break;
@@ -297,10 +298,22 @@ public class FileTxnLog implements TxnLog {
             }
         } catch (IOException e) {
             LOG.warn("Unexpected exception", e);
+        } finally {
+            close(itr);
         }
         return zxid;
     }
 
+    private void close(TxnIterator itr) {
+        if (itr != null) {
+            try {
+                itr.close();
+            } catch (IOException ioe) {
+                LOG.warn("Error closing file iterator", ioe);
+            }
+        }
+    }
+
     /**
      * commit the logs. make sure that evertyhing hits the
      * disk
@@ -361,17 +374,22 @@ public class FileTxnLog implements TxnLog {
      * @return true if successful false if not
      */
     public boolean truncate(long zxid) throws IOException {
-        FileTxnIterator itr = new FileTxnIterator(this.logDir, zxid);
-        PositionInputStream input = itr.inputStream;
-        long pos = input.getPosition();
-        // now, truncate at the current position
-        RandomAccessFile raf=new RandomAccessFile(itr.logFile,"rw");
-        raf.setLength(pos);
-        raf.close();
-        while(itr.goToNextLog()) {
-            if (!itr.logFile.delete()) {
-                LOG.warn("Unable to truncate " + itr.logFile);
+        FileTxnIterator itr = null;
+        try {
+            itr = new FileTxnIterator(this.logDir, zxid);
+            PositionInputStream input = itr.inputStream;
+            long pos = input.getPosition();
+            // now, truncate at the current position
+            RandomAccessFile raf=new RandomAccessFile(itr.logFile,"rw");
+            raf.setLength(pos);
+            raf.close();
+            while(itr.goToNextLog()) {
+                if (!itr.logFile.delete()) {
+                    LOG.warn("Unable to truncate {}", itr.logFile);
+                }
             }
+        } finally {
+            close(itr);
         }
         return true;
     }
@@ -700,7 +718,9 @@ public class FileTxnLog implements TxnLog {
          * and release the resources.
          */
         public void close() throws IOException {
-            inputStream.close();
+            if (inputStream != null) {
+                inputStream.close();
+            }
         }
     }
 

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

@@ -162,30 +162,36 @@ public class FileTxnSnapLog {
         TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
         long highestZxid = dt.lastProcessedZxid;
         TxnHeader hdr;
-        while (true) {
-            // iterator points to
-            // the first valid txn when initialized
-            hdr = itr.getHeader();
-            if (hdr == null) {
-                //empty logs
-                return dt.lastProcessedZxid;
+        try {
+            while (true) {
+                // iterator points to
+                // the first valid txn when initialized
+                hdr = itr.getHeader();
+                if (hdr == null) {
+                    //empty logs
+                    return dt.lastProcessedZxid;
+                }
+                if (hdr.getZxid() < highestZxid && highestZxid != 0) {
+                    LOG.error("{}(higestZxid) > {}(next log) for type {}",
+                            new Object[] { highestZxid, hdr.getZxid(),
+                                    hdr.getType() });
+                } else {
+                    highestZxid = hdr.getZxid();
+                }
+                try {
+                    processTransaction(hdr,dt,sessions, itr.getTxn());
+                } catch(KeeperException.NoNodeException e) {
+                   throw new IOException("Failed to process transaction type: " +
+                         hdr.getType() + " error: " + e.getMessage(), e);
+                }
+                listener.onTxnLoaded(hdr, itr.getTxn());
+                if (!itr.next())
+                    break;
             }
-            if (hdr.getZxid() < highestZxid && highestZxid != 0) {
-                LOG.error(highestZxid + "(higestZxid) > "
-                        + hdr.getZxid() + "(next log) for type "
-                        + hdr.getType());
-            } else {
-                highestZxid = hdr.getZxid();
+        } finally {
+            if (itr != null) {
+                itr.close();
             }
-            try {
-                processTransaction(hdr,dt,sessions, itr.getTxn());
-            } catch(KeeperException.NoNodeException e) {
-               throw new IOException("Failed to process transaction type: " +
-                     hdr.getType() + " error: " + e.getMessage(), e);
-            }
-            listener.onTxnLoaded(hdr, itr.getTxn());
-            if (!itr.next())
-                break;
         }
         return highestZxid;
     }

src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java

@@ -39,6 +39,7 @@ import org.apache.jute.Record;
 import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.TxnLogProposalIterator;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.server.quorum.Leader.Proposal;
@@ -729,6 +730,11 @@ public class LearnerHandler extends Thread {
                                                          null, maxCommittedLog);
                     needSnap = false;
                 }
+                // closing the resources
+                if (txnLogItr instanceof TxnLogProposalIterator) {
+                    TxnLogProposalIterator txnProposalItr = (TxnLogProposalIterator) txnLogItr;
+                    txnProposalItr.close();
+                }
             } else {
                 LOG.warn("Unhandled scenario for peer sid: " +  getSid());
             }

src/java/test/org/apache/zookeeper/server/quorum/LearnerHandlerTest.java

@@ -32,6 +32,7 @@ import java.util.Queue;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.TxnLogProposalIterator;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.Leader.Proposal;
@@ -315,6 +316,32 @@ public class LearnerHandlerTest extends ZKTestCase {
 
     }
 
+    /**
+     * Test case verifying TxnLogProposalIterator closure.
+     */
+    @Test
+    public void testTxnLogProposalIteratorClosure() throws Exception {
+        long peerZxid;
+
+        // CommmitedLog is empty, we will use txnlog up to lastProcessZxid
+        db = new MockZKDatabase(null) {
+            @Override
+            public Iterator<Proposal> getProposalsFromTxnLog(long peerZxid,
+                    long limit) {
+                return TxnLogProposalIterator.EMPTY_ITERATOR;
+            }
+        };
+        db.lastProcessedZxid = 7;
+        db.txnLog.add(createProposal(2));
+        db.txnLog.add(createProposal(3));
+
+        // Peer zxid
+        peerZxid = 4;
+        assertTrue("Couldn't identify snapshot transfer!",
+                learnerHandler.syncFollower(peerZxid, db, leader));
+        reset();
+    }
+
     /**
      * Test cases when txnlog is enabled and commitedLog is empty
      */

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

@@ -110,6 +110,8 @@ public class TruncateTest extends ZKTestCase {
         txn = iter.getTxn();
         Assert.assertEquals(200, hdr.getZxid());
         Assert.assertTrue(txn instanceof SetDataTxn);
+        iter.close();
+        ClientBase.recursiveDelete(tmpdir);
     }
 
     private void append(ZKDatabase zkdb, int i) throws IOException {