Merge branch 'cassandra-3.0' into trunk

test/unit/org/apache/cassandra/db/KeyCacheTest.java

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.db;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -24,6 +25,7 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Uninterruptibles;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -35,6 +37,7 @@ import org.apache.cassandra.cache.KeyCacheKey;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.exceptions.ConfigurationException;
@@ -50,6 +53,8 @@ public class KeyCacheTest
     private static final String KEYSPACE1 = "KeyCacheTest1";
     private static final String COLUMN_FAMILY1 = "Standard1";
     private static final String COLUMN_FAMILY2 = "Standard2";
+    private static final String COLUMN_FAMILY3 = "Standard3";
+
 
     @BeforeClass
     public static void defineSchema() throws ConfigurationException
@@ -58,7 +63,8 @@ public class KeyCacheTest
         SchemaLoader.createKeyspace(KEYSPACE1,
                                     KeyspaceParams.simple(1),
                                     SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY1),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY2));
+                                    SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY2),
+                                    SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY3));
     }
 
     @AfterClass
@@ -83,7 +89,7 @@ public class KeyCacheTest
         store.forceBlockingFlush();
 
         // populate the cache
-        readData(KEYSPACE1, COLUMN_FAMILY2, 100);
+        readData(KEYSPACE1, COLUMN_FAMILY2, 0, 100);
         assertKeyCacheSize(100, KEYSPACE1, COLUMN_FAMILY2);
 
         // really? our caches don't implement the map interface? (hence no .addAll)
@@ -119,6 +125,56 @@ public class KeyCacheTest
         }
     }
 
+    @Test
+    public void testKeyCacheLoadWithLostTable() throws Exception
+    {
+        CompactionManager.instance.disableAutoCompaction();
+
+        ColumnFamilyStore store = Keyspace.open(KEYSPACE1).getColumnFamilyStore(COLUMN_FAMILY3);
+
+        // empty the cache
+        CacheService.instance.invalidateKeyCache();
+        assertKeyCacheSize(0, KEYSPACE1, COLUMN_FAMILY3);
+
+        // insert data and force to disk
+        SchemaLoader.insertData(KEYSPACE1, COLUMN_FAMILY3, 0, 100);
+        store.forceBlockingFlush();
+
+        Collection<SSTableReader> firstFlushTables = ImmutableList.copyOf(store.getLiveSSTables());
+
+        // populate the cache
+        readData(KEYSPACE1, COLUMN_FAMILY3, 0, 100);
+        assertKeyCacheSize(100, KEYSPACE1, COLUMN_FAMILY3);
+
+        // insert some new data and force to disk
+        SchemaLoader.insertData(KEYSPACE1, COLUMN_FAMILY3, 100, 50);
+        store.forceBlockingFlush();
+
+        // check that it's fine
+        readData(KEYSPACE1, COLUMN_FAMILY3, 100, 50);
+        assertKeyCacheSize(150, KEYSPACE1, COLUMN_FAMILY3);
+
+        // force the cache to disk
+        CacheService.instance.keyCache.submitWrite(Integer.MAX_VALUE).get();
+
+        CacheService.instance.invalidateKeyCache();
+        assertKeyCacheSize(0, KEYSPACE1, COLUMN_FAMILY3);
+
+        // check that the content is written correctly
+        CacheService.instance.keyCache.loadSaved();
+        assertKeyCacheSize(150, KEYSPACE1, COLUMN_FAMILY3);
+
+        CacheService.instance.invalidateKeyCache();
+        assertKeyCacheSize(0, KEYSPACE1, COLUMN_FAMILY3);
+
+        // now remove the first sstable from the store to simulate losing the file
+        store.markObsolete(firstFlushTables, OperationType.UNKNOWN);
+
+        // check that reading now correctly skips over lost table and reads the rest (CASSANDRA-10219)
+        CacheService.instance.keyCache.loadSaved();
+        assertKeyCacheSize(50, KEYSPACE1, COLUMN_FAMILY3);
+    }
+
     @Test
     public void testKeyCache() throws ExecutionException, InterruptedException
     {
@@ -175,13 +231,11 @@ public class KeyCacheTest
         assertKeyCacheSize(noEarlyOpen ? 4 : 2, KEYSPACE1, COLUMN_FAMILY1);
     }
 
-    private static void readData(String keyspace, String columnFamily, int numberOfRows)
+    private static void readData(String keyspace, String columnFamily, int startRow, int numberOfRows)
     {
         ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(columnFamily);
-        CFMetaData cfm = Schema.instance.getCFMetaData(keyspace, columnFamily);
-
         for (int i = 0; i < numberOfRows; i++)
-            Util.getAll(Util.cmd(store, "key" + i).includeRow("col" + i).build());
+            Util.getAll(Util.cmd(store, "key" + (i + startRow)).includeRow("col" + (i + startRow)).build());
     }