Fixes for fully-expired SSTable index notification in compaction

src/java/org/apache/cassandra/db/compaction/CompactionTask.java

@@ -71,6 +71,7 @@
 import org.apache.cassandra.service.snapshot.SnapshotOptions;
 import org.apache.cassandra.service.snapshot.SnapshotType;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.TimeUUID;
 import org.apache.cassandra.utils.concurrent.Refs;
 
@@ -623,6 +624,10 @@ private void maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(Set<SSTableReade
 
         for (SSTableReader expiredSSTable : fullyExpiredSSTables)
         {
+            // Notifying indexers is best-effort: the SSTable is fully expired and is going to be obsoleted regardless.
+            // A read error (e.g. a corrupt-but-expired SSTable, which was previously dropped without ever being read) or
+            // a failure in a custom indexer must not silently leave the SSTable undropped, or every subsequent compaction
+            // would hit the same failure, leaving the expired SSTables permanently stuck.
             try (ISSTableScanner scanner = expiredSSTable.getScanner())
             {
                 while (scanner.hasNext())
@@ -649,22 +654,41 @@ private void maybeNotifyIndexersAboutRowsInFullyExpiredSSTables(Set<SSTableReade
                             for (Index.Indexer indexer : indexers)
                                 indexer.begin();
 
-                            while (partition.hasNext())
+                            try
                             {
-                                Unfiltered unfiltered = partition.next();
-                                if (unfiltered instanceof Row)
+                                Row staticRow = partition.staticRow();
+                                if (!staticRow.isEmpty())
                                 {
                                     for (Index.Indexer indexer : indexers)
-                                        indexer.removeRow((Row) unfiltered);
+                                        indexer.removeRow(staticRow);
                                 }
-                            }
 
-                            for (Index.Indexer indexer : indexers)
-                                indexer.finish();
+                                while (partition.hasNext())
+                                {
+                                    Unfiltered unfiltered = partition.next();
+                                    if (unfiltered instanceof Row)
+                                    {
+                                        for (Index.Indexer indexer : indexers)
+                                            indexer.removeRow((Row) unfiltered);
+                                    }
+                                }
+                            }
+                            finally
+                            {
+                                for (Index.Indexer indexer : indexers)
+                                    indexer.finish();
+                            }
                         }
                     }
                 }
             }
+            catch (Throwable t)
+            {
+                JVMStabilityInspector.inspectThrowable(t);
+                logger.warn("Failed to notify secondary indexes about rows in fully expired SSTable {} during compaction {}; " +
+                            "the SSTable will still be dropped, but index cleanup for it may be incomplete.",
+                            expiredSSTable, transaction.opIdString(), t);
+            }
         }
     }
 }

src/java/org/apache/cassandra/index/accord/NoOpIndex.java

@@ -118,6 +118,12 @@ public Indexer indexerFor(DecoratedKey key,
         return null;
     }
 
+    @Override
+    public boolean notifyIndexerAboutRowsInFullyExpiredSSTables()
+    {
+        return false;
+    }
+
     @Override
     public boolean supportsExpression(ColumnMetadata column, Operator operator)
     {

src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java

@@ -368,6 +368,12 @@ public void updateRow(Row oldRowData, Row newRowData)
         };
     }
 
+    @Override
+    public boolean notifyIndexerAboutRowsInFullyExpiredSSTables()
+    {
+        return false;
+    }
+
     @Override
     public boolean supportsExpression(ColumnMetadata column, Operator operator)
     {

src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java

@@ -254,6 +254,12 @@ public Indexer indexerFor(DecoratedKey key, RegularAndStaticColumns columns, lon
         return indexer;
     }
 
+    @Override
+    public boolean notifyIndexerAboutRowsInFullyExpiredSSTables()
+    {
+        return false;
+    }
+
     public Searcher searcherFor(ReadCommand command)
     {
         throw new UnsupportedOperationException();

test/unit/org/apache/cassandra/index/CustomIndexTest.java

@@ -744,6 +744,72 @@ public void notifyIndexesOfFullyExpiredSSTablesDuringCompaction()
         Assert.assertEquals(3, index2.rowsDeleted.size());
     }
 
+    @Test
+    public void indexerFailureDoesNotBlockDroppingFullyExpiredSSTables()
+    {
+        createTable("CREATE TABLE %s (id int primary key, col1 int) " +
+                    "WITH compaction = {'class': 'TimeWindowCompactionStrategy', " +
+                    "                   'compaction_window_size': 1," +
+                    "                   'compaction_window_unit': 'MINUTES'," +
+                    "                   'expired_sstable_check_frequency_seconds': 10} " +
+                    "AND gc_grace_seconds = 0");
+
+        createIndex(String.format("CREATE CUSTOM INDEX throwing_expired_index ON %%s(col1) USING '%s'", ThrowingStubIndex.class.getName()));
+
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+
+        execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 0, 0);
+        execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 1, 1);
+        execute("INSERT INTO %s (id, col1) VALUES (?, ?) USING TTL 20", 2, 2);
+
+        flush();
+        Assert.assertFalse(cfs.getLiveSSTables().isEmpty());
+
+        // Let the rows (and SSTable) fully expire.
+        Uninterruptibles.sleepUninterruptibly(60, TimeUnit.SECONDS);
+
+        // The indexer throws while being notified, but compaction must complete and still drop the expired SSTables.
+        compact();
+
+        Assert.assertTrue("Fully expired SSTables should have been dropped despite the indexer throwing",
+                          cfs.getLiveSSTables().isEmpty());
+    }
+
+    @Test
+    public void notifyIndexesOfStaticRowsInFullyExpiredSSTablesDuringCompaction()
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, s int static, v int, PRIMARY KEY (pk, ck)) " +
+                    "WITH compaction = {'class': 'TimeWindowCompactionStrategy', " +
+                    "                   'compaction_window_size': 1," +
+                    "                   'compaction_window_unit': 'MINUTES'," +
+                    "                   'expired_sstable_check_frequency_seconds': 10} " +
+                    "AND gc_grace_seconds = 0");
+
+        createIndex(String.format("CREATE CUSTOM INDEX static_expired_index ON %%s(s) USING '%s'", StubIndex.class.getName()));
+
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        StubIndex index = (StubIndex) cfs.indexManager.getIndexByName("static_expired_index");
+        Assert.assertNotNull(index);
+
+        // Static-only writes: each partition gets a static row and no clustering rows, so removeRow would never be
+        // invoked for these partitions unless the static row is handled explicitly.
+        execute("INSERT INTO %s (pk, s) VALUES (?, ?) USING TTL 20", 0, 100);
+        execute("INSERT INTO %s (pk, s) VALUES (?, ?) USING TTL 20", 1, 200);
+
+        flush();
+        Assert.assertFalse(cfs.getLiveSSTables().isEmpty());
+
+        Uninterruptibles.sleepUninterruptibly(60, TimeUnit.SECONDS);
+
+        // Ignore anything recorded during index build/flush; only care about the compaction-time notifications.
+        index.reset();
+        compact();
+
+        Assert.assertEquals("Static rows in fully-expired SSTables should be forwarded to the indexer",
+                            2, index.rowsDeleted.size());
+        Assert.assertTrue(cfs.getLiveSSTables().isEmpty());
+    }
+
     @Test
     public void validateOptions()
     {
@@ -1020,6 +1086,39 @@ private static IndexTarget indexTarget(String name, IndexTarget.Type type)
         return new IndexTarget(ColumnIdentifier.getInterned(name, true), type);
     }
 
+    // A stub index whose Indexer throws while removing rows, to simulate a misbehaving custom index (or a read error)
+    // while being notified about rows in a fully-expired SSTable during compaction.
+    public static class ThrowingStubIndex extends StubIndex
+    {
+        public ThrowingStubIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
+        {
+            super(baseCfs, metadata);
+        }
+
+        @Override
+        public Indexer indexerFor(DecoratedKey key,
+                                  RegularAndStaticColumns columns,
+                                  long nowInSec,
+                                  WriteContext ctx,
+                                  IndexTransaction.Type transactionType,
+                                  Memtable memtable)
+        {
+            return new Indexer()
+            {
+                public void begin() { }
+                public void partitionDelete(DeletionTime deletionTime) { }
+                public void rangeTombstone(RangeTombstone tombstone) { }
+                public void insertRow(Row row) { }
+                public void updateRow(Row oldRowData, Row newRowData) { }
+                public void removeRow(Row row)
+                {
+                    throw new RuntimeException("simulated indexer failure during fully-expired SSTable notification");
+                }
+                public void finish() { }
+            };
+        }
+    }
+
     public static final class CountMetadataReloadsIndex extends StubIndex
     {
         private final AtomicInteger reloads = new AtomicInteger(0);