ZOOKEEPER-2332: Fix server failed to start for empty txn log

Reviewers: kezhuw, anmolnar, kezhuw
Author: fanyang89
Closes #2141 from fanyang89/ZOOKEEPER-2332

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

@@ -227,6 +227,13 @@ public class FileTxnLog implements TxnLog, Closeable {
         return prevLogsRunningTotal + getCurrentLogSize();
     }
 
+    /**
+     * Get log size limit
+     */
+    public static long getTxnLogSizeLimit() {
+        return txnLogSizeLimit;
+    }
+
     /**
      * creates a checksum algorithm to be used
      * @return the checksum used for this txnlog
@@ -701,14 +708,29 @@ public class FileTxnLog implements TxnLog, Closeable {
 
         /**
          * go to the next logfile
+         *
          * @return true if there is one and false if there is no
          * new file to be read
-         * @throws IOException
          */
         private boolean goToNextLog() throws IOException {
-            if (storedFiles.size() > 0) {
+            if (!storedFiles.isEmpty()) {
                 this.logFile = storedFiles.remove(storedFiles.size() - 1);
-                ia = createInputArchive(this.logFile);
+                try {
+                    ia = createInputArchive(this.logFile);
+                } catch (EOFException ex) {
+                    // If this file is the last log file in the database and is empty,
+                    // it means that the last time the file was created
+                    // before the header was written.
+                    if (storedFiles.isEmpty() && this.logFile.length() == 0) {
+                        boolean deleted = this.logFile.delete();
+                        if (!deleted) {
+                            throw new IOException("Failed to delete empty tail log file: " + this.logFile.getName());
+                        }
+                        LOG.warn("Delete empty tail log file to recover from corruption file: {}", this.logFile.getName());
+                        return false;
+                    }
+                    throw ex;
+                }
                 return true;
             }
             return false;

zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java

@@ -25,11 +25,17 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
+import java.io.EOFException;
 import java.io.File;
 import java.io.IOException;
+import java.io.PrintWriter;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
 import java.util.Random;
+import java.util.stream.Collectors;
 import org.apache.jute.Record;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.DummyWatcher;
@@ -296,6 +302,84 @@ public class FileTxnLogTest extends ZKTestCase {
         }
     }
 
+    private void prepareTxnLogs(File dir, int n) throws IOException {
+        FileTxnLog.setTxnLogSizeLimit(1);
+        FileTxnLog log = new FileTxnLog(dir);
+        CreateRequest record = new CreateRequest(null, new byte[NODE_SIZE],
+                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT.toFlag());
+        int zxid = 1;
+        for (int i = 0; i < n; i++) {
+            log.append(new Request(0, 0, 0, new TxnHeader(0, 0, zxid, 0, -1), record, zxid));
+            zxid++;
+            log.commit();
+        }
+        log.close();
+    }
+
+    @Test
+    public void testEmptyTailTxnLog() throws IOException {
+        long limit = FileTxnLog.getTxnLogSizeLimit();
+
+        // prepare a database with logs
+        File tmpDir = ClientBase.createTmpDir();
+        ClientBase.setupTestEnv();
+        prepareTxnLogs(tmpDir, 4);
+
+        // find the tail log and clear
+        List<File> files = Arrays.
+                stream(Objects.requireNonNull(tmpDir.listFiles((File f, String name) -> name.startsWith("log.")))).
+                sorted(Comparator.comparing(File::getName)).
+                collect(Collectors.toList());
+        File toClear = files.get(files.size() - 1);
+        PrintWriter writer = new PrintWriter(toClear);
+        writer.close();
+        LOG.info("Clear the tail log file {}", toClear.getName());
+
+        // open txn log and iterate
+        try {
+            FileTxnLog.FileTxnIterator itr = new FileTxnLog.FileTxnIterator(tmpDir, 0x0, false);
+            while (itr.next()) {}
+        } catch (EOFException ex) {}
+
+        FileTxnLog.FileTxnIterator itr = new FileTxnLog.FileTxnIterator(tmpDir, 0x0, false);
+        while (itr.next()) {}
+
+        FileTxnLog.setTxnLogSizeLimit(limit);
+    }
+
+    @Test
+    public void testEmptyMedianTxnLog() throws IOException {
+        long limit = FileTxnLog.getTxnLogSizeLimit();
+
+        // prepare a database with logs
+        File tmpDir = ClientBase.createTmpDir();
+        ClientBase.setupTestEnv();
+        prepareTxnLogs(tmpDir, 4);
+
+        // find the median log and clear
+        List<File> files = Arrays.
+                stream(Objects.requireNonNull(tmpDir.listFiles((File f, String name) -> name.startsWith("log.")))).
+                sorted(Comparator.comparing(File::getName)).
+                collect(Collectors.toList());
+        File toClear = files.get(files.size() - 2);
+
+        PrintWriter writer = new PrintWriter(toClear);
+        writer.close();
+        LOG.info("Clear the median log file {}", toClear.getName());
+
+        // open txn log and iterate, should throw EOFException
+        boolean isEof = false;
+        try {
+            FileTxnLog.FileTxnIterator itr = new FileTxnLog.FileTxnIterator(tmpDir, 0x0, false);
+            while (itr.next()) {}
+        } catch (EOFException ex) {
+            isEof = true;
+        }
+        assertTrue(isEof, "Median txn log file empty should throw Exception");
+
+        FileTxnLog.setTxnLogSizeLimit(limit);
+    }
+
     private int calculateSingleRecordLength(TxnHeader txnHeader, Record record) throws IOException {
         int crcLength = 8;
         int dataLength = 4;