ZOOKEEPER-2325: Data inconsistency if all snapshots empty or missing

Author: Benjamin Reed <br33d@fb.com>
Author: Benjamin Reed <breed@apache.org>

Reviewers: Raúl Gutiérrez Segalés <rgs@apache.org>, Michael Han <hanm@apache.org>, Allan Lyu <lvfangmin@gmail.com>

Closes #117 from breed/ZOOKEEPER-2325 and squashes the following commits:

35ce7e2 [Benjamin Reed] address comments from rgs1
68433ae [Benjamin Reed] ZOOKEEPER-2325: Data inconsistency if all snapshots empty or missing

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

@@ -165,8 +165,22 @@ public class FileTxnSnapLog {
      */
     public long restore(DataTree dt, Map<Long, Integer> sessions,
             PlayBackListener listener) throws IOException {
-        snapLog.deserialize(dt, sessions);
+        long deserializeResult = snapLog.deserialize(dt, sessions);
         FileTxnLog txnLog = new FileTxnLog(dataDir);
+        if (-1L == deserializeResult) {
+            /* this means that we couldn't find any snapshot, so we need to
+             * initialize an empty database (reported in ZOOKEEPER-2325) */
+            if (txnLog.getLastLoggedZxid() != -1) {
+                throw new IOException(
+                        "No snapshot found, but there are log entries. " +
+                        "Something is broken!");
+            }
+            /* TODO: (br33d) we should either put a ConcurrentHashMap on restore()
+             *       or use Map on save() */
+            save(dt, (ConcurrentHashMap<Long, Integer>)sessions);
+            /* return a zxid of zero, since we the database is empty */
+            return 0;
+        }
         TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
         long highestZxid = dt.lastProcessedZxid;
         TxnHeader hdr;

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

@@ -34,6 +34,7 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.Map;
 
 import org.apache.jute.BinaryInputArchive;
@@ -48,6 +49,7 @@ import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.server.ByteBufferInputStream;
 import org.apache.zookeeper.server.ByteBufferOutputStream;
+import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ServerCnxn;
 import org.apache.zookeeper.server.ServerCnxnFactory;
@@ -1336,6 +1338,7 @@ public class Zab1_0Test extends ZKTestCase {
             FileTxnSnapLog logFactory = new FileTxnSnapLog(tmpDir, tmpDir);
             File version2 = new File(tmpDir, "version-2");
             version2.mkdir();
+            logFactory.save(new DataTree(), new ConcurrentHashMap<Long, Integer>());
             long zxid = ZxidUtils.makeZxid(3, 3);
             logFactory.append(new Request(1, 1, ZooDefs.OpCode.error,
                     new TxnHeader(1, 1, zxid, 1, ZooDefs.OpCode.error),

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

@@ -0,0 +1,133 @@
+/**
+ * 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.test;
+
+import java.io.IOException;
+import java.io.File;
+import java.io.PrintWriter;
+import java.util.List;
+import java.util.LinkedList;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.server.quorum.Leader.Proposal;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.SyncRequestProcessor;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.junit.Assert;
+import org.junit.Test;
+
+/** If snapshots are corrupted to the empty file or deleted, Zookeeper should 
+ *  not proceed to read its transactiong log files
+ *  Test that zxid == -1 in the presence of emptied/deleted snapshots
+ */
+public class EmptiedSnapshotRecoveryTest extends ZKTestCase implements  Watcher {
+    private static final Logger LOG = Logger.getLogger(RestoreCommittedLogTest.class);
+    private static String HOSTPORT = "127.0.0.1:" + PortAssignment.unique();
+    private static final int CONNECTION_TIMEOUT = 3000;
+    private static final int N_TRANSACTIONS = 150;
+    private static final int SNAP_COUNT = 100;
+
+    public void runTest(boolean leaveEmptyFile) throws Exception {
+        File tmpSnapDir = ClientBase.createTmpDir();
+        File tmpLogDir  = ClientBase.createTmpDir();
+        ClientBase.setupTestEnv();
+        ZooKeeperServer zks = new ZooKeeperServer(tmpSnapDir, tmpLogDir, 3000);
+        SyncRequestProcessor.setSnapCount(SNAP_COUNT);
+        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
+        ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
+        f.startup(zks);
+        Assert.assertTrue("waiting for server being up ",
+                ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));
+        ZooKeeper zk = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this);
+        try {
+            for (int i = 0; i< N_TRANSACTIONS; i++) {
+                zk.create("/node-" + i, new byte[0], Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT);
+            }
+        } finally {
+            zk.close();
+        }
+        f.shutdown();
+        zks.shutdown();
+        Assert.assertTrue("waiting for server to shutdown",
+                ClientBase.waitForServerDown(HOSTPORT, CONNECTION_TIMEOUT));
+
+        // start server again with intact database
+        zks = new ZooKeeperServer(tmpSnapDir, tmpLogDir, 3000);
+        zks.startdata();
+        long zxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
+        LOG.info("After clean restart, zxid = " + zxid);
+        Assert.assertTrue("zxid > 0", zxid > 0);
+        zks.shutdown();
+
+        // Make all snapshots empty
+        FileTxnSnapLog txnLogFactory = zks.getTxnLogFactory();
+        List<File> snapshots = txnLogFactory.findNRecentSnapshots(10);
+        Assert.assertTrue("We have a snapshot to corrupt", snapshots.size() > 0);
+        for (File file: snapshots) {
+            if (leaveEmptyFile) {
+                new PrintWriter(file).close ();
+            } else {
+                file.delete();
+            }
+        }
+
+        // start server again with corrupted database
+        zks = new ZooKeeperServer(tmpSnapDir, tmpLogDir, 3000);
+        try {
+            zks.startdata();
+            zxid = zks.getZKDatabase().loadDataBase();
+            Assert.fail("Should have gotten exception for corrupted database");
+        } catch (IOException e) {
+            // expected behavior
+        } 
+        zks.shutdown();
+    }
+
+    /**
+     * Test resilience to empty Snapshots
+     * @throws Exception an exception might be thrown here
+     */
+    @Test
+    public void testRestoreWithEmptySnapFiles() throws Exception {
+        runTest(true);
+    }
+
+    /**
+     * Test resilience to deletion of Snapshots
+     * @throws Exception an exception might be thrown here
+     */
+    @Test
+    public void testRestoreWithNoSnapFiles() throws Exception {
+        runTest(false);
+    }
+
+    public void process(WatchedEvent event) {
+        // do nothing
+    }
+
+}

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

@@ -72,6 +72,9 @@ public class TruncateTest extends ZKTestCase {
         File tmpdir = ClientBase.createTmpDir();
         FileTxnSnapLog snaplog = new FileTxnSnapLog(tmpdir, tmpdir);
         ZKDatabase zkdb = new ZKDatabase(snaplog);
+        // make sure to snapshot, so that we have something there when
+        // truncateLog reloads the db
+        snaplog.save(zkdb.getDataTree(), zkdb.getSessionWithTimeOuts());
 
         for (int i = 1; i <= 100; i++) {
             append(zkdb, i);