Don't throw assertion error on old key cache keys just skip serializing them.

patch by aweisberg; reviewed by slebresne for CASSANDRA-10778

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.1
+ * Fix error when saving cached key for old format sstable (CASSANDRA-10778)
  * Invalidate prepared statements on DROP INDEX (CASSANDRA-10758)
  * Fix SELECT statement with IN restrictions on partition key,
    ORDER BY and LIMIT (CASSANDRA-10729)

src/java/org/apache/cassandra/service/CacheService.java

@@ -455,6 +455,10 @@ public class CacheService implements CacheServiceMBean
     {
         public void serialize(KeyCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException
         {
+            //Don't serialize old format entries since we didn't bother to implement serialization of both for simplicity
+            //https://issues.apache.org/jira/browse/CASSANDRA-10778
+            if (!key.desc.version.storeRows()) return;
+
             RowIndexEntry entry = CacheService.instance.keyCache.getInternal(key);
             if (entry == null)
                 return;

test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java

@@ -56,6 +56,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.Version;
 import org.apache.cassandra.io.sstable.format.big.BigFormat;
 import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.service.CacheService;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.streaming.StreamPlan;
 import org.apache.cassandra.streaming.StreamSession;
@@ -238,7 +239,7 @@ public class LegacySSTableTest
         loadLegacyTables();
     }
 
-    private static void loadLegacyTables() throws IOException
+    private static void loadLegacyTables() throws Exception
     {
         for (String legacyVersion : legacyVersions)
         {
@@ -251,6 +252,8 @@ public class LegacySSTableTest
             loadLegacyTable("legacy_%s_clust", legacyVersion);
             loadLegacyTable("legacy_%s_clust_counter", legacyVersion);
 
+            CacheService.instance.invalidateKeyCache();
+            long startCount = CacheService.instance.keyCache.size();
             for (int ck = 0; ck < 50; ck++)
             {
                 String ckValue = Integer.toString(ck) + longString;
@@ -286,6 +289,20 @@ public class LegacySSTableTest
                     Assert.assertEquals(1L, rs.one().getLong("val"));
                 }
             }
+
+            //For https://issues.apache.org/jira/browse/CASSANDRA-10778
+            //Validate whether the key cache successfully saves in the presence of old keys as
+            //well as loads the correct number of keys
+            long endCount = CacheService.instance.keyCache.size();
+            Assert.assertTrue(endCount > startCount);
+            CacheService.instance.keyCache.submitWrite(Integer.MAX_VALUE).get();
+            CacheService.instance.invalidateKeyCache();
+            Assert.assertEquals(startCount, CacheService.instance.keyCache.size());
+            CacheService.instance.keyCache.loadSaved();
+            if (BigFormat.instance.getVersion(legacyVersion).storeRows())
+                Assert.assertEquals(endCount, CacheService.instance.keyCache.size());
+            else
+                Assert.assertEquals(startCount, CacheService.instance.keyCache.size());
         }
     }