ZOOKEEPER-683. LogFormatter fails to parse transactional log files (phunt via mahadev)

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

CHANGES.txt

@@ -234,6 +234,9 @@ BUGFIXES:
   ZOOKEEPER-669. watchedevent tostring should clearly output the
   state/type/path (phunt via mahadev)
 
+  ZOOKEEPER-683. LogFormatter fails to parse transactional log files (phunt
+  via 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/LogFormatter.java

@@ -21,13 +21,19 @@ package org.apache.zookeeper.server;
 import java.io.ByteArrayInputStream;
 import java.io.EOFException;
 import java.io.FileInputStream;
+import java.io.IOException;
 import java.text.DateFormat;
 import java.util.Date;
-
-import org.apache.log4j.Logger;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
 
 import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.InputArchive;
+import org.apache.jute.Record;
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.server.persistence.FileHeader;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.txn.TxnHeader;
 
 public class LogFormatter {
@@ -43,23 +49,51 @@ public class LogFormatter {
         }
         FileInputStream fis = new FileInputStream(args[0]);
         BinaryInputArchive logStream = BinaryInputArchive.getArchive(fis);
+        FileHeader fhdr = new FileHeader();
+        fhdr.deserialize(logStream, "fileheader");
+
+        if (fhdr.getMagic() != FileTxnLog.TXNLOG_MAGIC) {
+            System.err.println("Invalid magic number for " + args[0]);
+            System.exit(2);
+        }
+        System.out.println("ZooKeeper Transactional Log File with dbid "
+                + fhdr.getDbid() + " txnlog format version "
+                + fhdr.getVersion());
+
+        int count = 0;
         while (true) {
-            byte[] bytes = logStream.readBuffer("txnEntry");
+            long crcValue;
+            byte[] bytes;
+            try {
+                crcValue = logStream.readLong("crcvalue");
+
+                bytes = logStream.readBuffer("txnEntry");
+            } catch (EOFException e) {
+                System.out.println("EOF reached after " + count + " txns.");
+                return;
+            }
             if (bytes.length == 0) {
                 // Since we preallocate, we define EOF to be an
                 // empty transaction
-                throw new EOFException();
+                System.out.println("EOF reached after " + count + " txns.");
+                return;
+            }
+            Checksum crc = new Adler32();
+            crc.update(bytes, 0, bytes.length);
+            if (crcValue != crc.getValue()) {
+                throw new IOException("CRC doesn't match " + crcValue +
+                        " vs " + crc.getValue());
             }
-            InputArchive ia = BinaryInputArchive
-                    .getArchive(new ByteArrayInputStream(bytes));
+            InputArchive iab = BinaryInputArchive
+                                .getArchive(new ByteArrayInputStream(bytes));
             TxnHeader hdr = new TxnHeader();
-            hdr.deserialize(ia, "hdr");
+            SerializeUtils.deserializeTxn(iab, hdr);
             System.out.println(DateFormat.getDateTimeInstance(DateFormat.SHORT,
                     DateFormat.LONG).format(new Date(hdr.getTime()))
                     + " session 0x"
                     + Long.toHexString(hdr.getClientId())
-                    + ":"
-                    + hdr.getCxid()
+                    + " cxid 0x"
+                    + Long.toHexString(hdr.getCxid())
                     + " zxid 0x"
                     + Long.toHexString(hdr.getZxid())
                     + " " + TraceFormatter.op2String(hdr.getType()));
@@ -67,6 +101,7 @@ public class LogFormatter {
                 LOG.error("Last transaction was partial.");
                 throw new EOFException("Last transaction was partial.");
             }
+            count++;
         }
     }
 }

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

@@ -47,7 +47,44 @@ import org.apache.zookeeper.txn.TxnHeader;
 /**
  * This class implements the TxnLog interface. It provides api's
  * to access the txnlogs and add entries to it.
- *
+ * <p>
+ * The format of a Transactional log is as follows:
+ * <blockquote><pre>
+ * LogFile:
+ *     FileHeader TxnList ZeroPad
+ * 
+ * FileHeader: {
+ *     magic 4bytes (ZKLG)
+ *     version 4bytes
+ *     dbid 8bytes
+ *   }
+ * 
+ * TxnList:
+ *     Txn || Txn TxnList
+ *     
+ * Txn:
+ *     checksum Txnlen TxnHeader Record 0x42
+ * 
+ * checksum: 8bytes Adler32 is currently used
+ *   calculated across payload -- Txnlen, TxnHeader, Record and 0x42
+ * 
+ * Txnlen:
+ *     len 4bytes
+ * 
+ * TxnHeader: {
+ *     sessionid 8bytes
+ *     cxid 4bytes
+ *     zxid 8bytes
+ *     time 8bytes
+ *     type 4bytes
+ *   }
+ *     
+ * Record:
+ *     See Jute definition file for details on the various record types
+ *      
+ * ZeroPad:
+ *     0 padded to EOF (filled during preallocation stage)
+ * </pre></blockquote> 
  */
 public class FileTxnLog implements TxnLog {
     private static final Logger LOG;

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

@@ -31,9 +31,11 @@ import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.server.LogFormatter;
 import org.apache.zookeeper.server.NIOServerCnxn;
 import org.apache.zookeeper.server.SyncRequestProcessor;
 import org.apache.zookeeper.server.ZooKeeperServer;
+import org.junit.Test;
 
 public class InvalidSnapshotTest extends TestCase implements Watcher {
     private final static Logger LOG = Logger.getLogger(UpgradeTest.class);
@@ -53,10 +55,22 @@ public class InvalidSnapshotTest extends TestCase implements Watcher {
         LOG.info("FINISHED " + getName());
     }
 
+    /**
+     * Verify the LogFormatter by running it on a known file.
+     */
+    @Test
+    public void testLogFormatter() throws Exception {
+        File snapDir = new File(testData, "invalidsnap");
+        File logfile = new File(new File(snapDir, "version-2"), "log.274");
+        String[] args = {logfile.getCanonicalFile().toString()};
+        LogFormatter.main(args);
+    }
+    
     /**
      * test the snapshot
      * @throws Exception an exception could be expected
      */
+    @Test
     public void testSnapshot() throws Exception {
         File snapDir = new File(testData, "invalidsnap");
         ZooKeeperServer zks = new ZooKeeperServer(snapDir, snapDir, 3000);