fix possible data loss during multiple restarts. patch by jbellis; reviewed by Jun Rao for CASSANDRA-204

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.3@779770 13f79535-47bb-0310-9956-ffa450edef68

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -188,7 +188,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         ssTables_.addAll(filenames);
         /* Load the index files and the Bloom Filters associated with them. */
         SSTable.onStart(filenames);
-        logger_.debug("Submitting a major compaction task ...");
         MinorCompactionManager.instance().submit(ColumnFamilyStore.this);
         if (columnFamily_.equals(Table.hints_))
         {
@@ -417,13 +416,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return table_ + "-" + columnFamily_ + "-" + SSTable.temporaryFile_ + "-" + index;
     }
 
-
-    /*
-    * This version is used only on start up when we are recovering from logs.
-    * In the future we may want to parellelize the log processing for a table
-    * by having a thread per log file present for recovery. Re-visit at that
-    * time.
-    */
     void switchMemtable()
     {
         memtableLock_.writeLock().lock();
@@ -445,12 +437,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         memtableSwitchCount++;
     }
 
-    /*
-     * This version is used only on start up when we are recovering from logs.
-     * In the future we may want to parellelize the log processing for a table
-     * by having a thread per log file present for recovery. Re-visit at that
-     * time.
-     */
     void switchBinaryMemtable(String key, byte[] buffer) throws IOException
     {
         binaryMemtable_.set(new BinaryMemtable(table_, columnFamily_));
@@ -481,7 +467,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
     void forceFlushBinary()
     {
         BinaryMemtableManager.instance().submit(getColumnFamilyName(), binaryMemtable_.get());
-        //binaryMemtable_.get().flush(true);
     }
 
     /**
@@ -728,7 +713,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
      */
     void applyNow(String key, ColumnFamily columnFamily) throws IOException
     {
-        getMemtableThreadSafe().putOnRecovery(key, columnFamily);
+        getMemtableThreadSafe().put(key, columnFamily);
     }
 
     /*

src/java/org/apache/cassandra/db/CommitLog.java

@@ -34,6 +34,8 @@ import org.apache.log4j.Logger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.apache.commons.lang.StringUtils;
+
 /*
  * Commit Log tracks every write operation into the system. The aim
  * of the commit log is to be able to successfully recover data that was
@@ -300,16 +302,12 @@ public class CommitLog
             file = clogs.get(i);
             readCommitLogHeader(file.getAbsolutePath(), header2);
             byte[] result = CommitLogHeader.and(header, header2);
-            if ( !CommitLogHeader.isZero(result) )
-            {
-                filesNeeded.push(file);
-            }
-            else
-            {
+            if (CommitLogHeader.isZero(result))
                 break;
-            }
+            filesNeeded.push(file);
         }
 
+        logger_.info("Replaying logs from " + StringUtils.join(filesNeeded, ", "));
         doRecovery(filesNeeded, header);
     }
 

src/java/org/apache/cassandra/db/Memtable.java

@@ -160,6 +160,8 @@ public class Memtable implements Comparable<Memtable>
         }
     }
 
+    /** flush synchronously (in the current thread, not on the executor).
+     *  only the recover code should call this. */
     void flushOnRecovery() throws IOException {
         if (!isClean())
             flush(CommitLog.CommitLogContext.NULL);
@@ -201,16 +203,6 @@ public class Memtable implements Comparable<Memtable>
         return builder.toString();
     }
 
-    /**
-     * This version is called on commit log recovery. The threshold
-     * is not respected and a forceFlush() needs to be invoked to flush
-     * the contents to disk.  Does not go through the executor.
-    */
-    void putOnRecovery(String key, ColumnFamily columnFamily)
-    {
-        resolve(key, columnFamily);
-    }
-
     ColumnFamily getLocalCopy(String key, String columnFamilyColumn, IFilter filter)
     {
     	String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn);

src/java/org/apache/cassandra/db/PeriodicFlushManager.java

@@ -76,7 +76,6 @@ class PeriodicFlushManager implements IComponentShutdown
                 columnFamilyStore.forceFlush();
             }
         };
-        logger_.info("start periodic flush daemon every " + flushPeriodInMinutes + " minutes for " + columnFamilyStore.columnFamily_);
         flusher_.scheduleWithFixedDelay(runnable, flushPeriodInMinutes, flushPeriodInMinutes, TimeUnit.MINUTES);       
     }
 }

src/java/org/apache/cassandra/io/SSTable.java

@@ -403,6 +403,7 @@ public class SSTable
 
     private void loadIndexFile() throws IOException
     {
+        logger_.debug("Loading indexes from " + dataFile_);
         IFileReader indexReader = null;
         /* Read all block indexes to maintain an index in memory */
         try
@@ -441,7 +442,6 @@ public class SSTable
                     String blockIndexKey = bufIn.readUTF();
                     if (!blockIndexKey.equals(SSTable.blockIndexKey_))
                     {
-                        logger_.debug(" Done reading the block indexes, Index has been created");
                         break;
                     }
                     /* read the size of the block index */