CASSANDRA-21527: Make runWithCompactionsDisabled return non-null on success (4.0)

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

@@ -90,6 +90,7 @@
 import org.apache.cassandra.utils.memory.MemtableAllocator;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;
+import org.apache.cassandra.exceptions.TruncateException;
 
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
@@ -1553,7 +1554,15 @@ public boolean rebuildOnFailedScrub(Throwable failure)
         if (!isIndex() || !SecondaryIndexManager.isIndexColumnFamilyStore(this))
             return false;
 
-        truncateBlocking();
+        try
+        {
+            truncateBlocking();
+        }
+        catch (TruncateException e)
+        {
+            logger.warn("Unable to truncate {} to rebuild index after scrub failure", name, e);
+            return false;
+        }
 
         logger.warn("Rebuilding index for {} because of <{}>", name, failure.getMessage());
 
@@ -1664,8 +1673,15 @@ public CleanupSummary releaseRepairData(Collection<UUID> sessions, boolean force
                 UUID session = sst.getPendingRepair();
                 return session != null && sessions.contains(session);
             };
-            return runWithCompactionsDisabled(() -> compactionStrategyManager.releaseRepairData(sessions),
-                                              predicate, false, true, true);
+            CleanupSummary summary = runWithCompactionsDisabled(() -> compactionStrategyManager.releaseRepairData(sessions),
+                                                                predicate, false, true, true);
+            if (summary == null)
+            {
+                logger.warn("Unable to cancel in-progress compactions for {}.{}, could not force release repair data for sessions {}",
+                            keyspace.getName(), name, sessions);
+                return new CleanupSummary(this, Collections.emptySet(), new HashSet<>(sessions));
+            }
+            return summary;
         }
         else
         {
@@ -2328,39 +2344,56 @@ private void truncateBlocking(boolean noSnapshot)
             for (SSTableReader sstable : cfs.getLiveSSTables())
                 now = Math.max(now, sstable.maxDataAge);
         truncatedAt = now;
-
-        Runnable truncateRunnable = new Runnable()
+        Throwable failure = null;
+        try
         {
-            public void run()
-            {
-                logger.info("Truncating {}.{} with truncatedAt={}", keyspace.getName(), getTableName(), truncatedAt);
-                // since truncation can happen at different times on different nodes, we need to make sure
-                // that any repairs are aborted, otherwise we might clear the data on one node and then
-                // stream in data that is actually supposed to have been deleted
-                ActiveRepairService.instance.abort((prs) -> prs.getTableIds().contains(metadata.id),
-                                                   "Stopping parent sessions {} due to truncation of tableId="+metadata.id);
-                data.notifyTruncated(truncatedAt);
+            Boolean succeeded = runWithCompactionsDisabled(() -> runTruncate(truncatedAt, noSnapshot, replayAfter),
+                                                             true, true);
+            // null means compactions couldn't be disabled, not failure of the truncate work itself
+            if (succeeded == null)
+                failure = new TruncateException("Unable to stop compaction. Usually retrying truncate will work.");
+        }
+        catch (Throwable t)
+        {
+            failure = t;
+        }
 
-            if (!noSnapshot && DatabaseDescriptor.isAutoSnapshot())
-                snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(name, SNAPSHOT_TRUNCATE_PREFIX));
+        try
+        {
+            viewManager.build();
+        }
+        catch (Throwable t)
+        {
+            failure = merge(failure, t);
+        }
 
-            discardSSTables(truncatedAt);
+        maybeFail(failure);
 
-            indexManager.truncateAllIndexesBlocking(truncatedAt);
-            viewManager.truncateBlocking(replayAfter, truncatedAt);
+        logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+    }
 
-                SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter);
-                logger.trace("cleaning out row cache");
-                invalidateCaches();
+    private boolean runTruncate(long truncatedAt, boolean noSnapshot, CommitLogPosition replayAfter)
+    {
+        logger.info("Truncating {}.{} with truncatedAt={}", keyspace.getName(), getTableName(), truncatedAt);
+        // since truncation can happen at different times on different nodes, we need to make sure
+        // that any repairs are aborted, otherwise we might clear the data on one node and then
+        // stream in data that is actually supposed to have been deleted
+        ActiveRepairService.instance.abort((prs) -> prs.getTableIds().contains(metadata.id),
+                                           "Stopping parent sessions {} due to truncation of tableId="+metadata.id);
+        data.notifyTruncated(truncatedAt);
 
-            }
-        };
+        if (!noSnapshot && DatabaseDescriptor.isAutoSnapshot())
+            snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(name, SNAPSHOT_TRUNCATE_PREFIX));
 
-        runWithCompactionsDisabled(Executors.callable(truncateRunnable), true, true);
+        discardSSTables(truncatedAt);
 
-        viewManager.build();
+        indexManager.truncateAllIndexesBlocking(truncatedAt);
+        viewManager.truncateBlocking(replayAfter, truncatedAt);
 
-        logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+        SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter);
+        logger.trace("cleaning out row cache");
+        invalidateCaches();
+        return true;
     }
 
     /**

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

@@ -942,7 +942,7 @@ public List<Future<?>> submitMaximal(final ColumnFamilyStore cfStore, final int
         // for ourselves to finish/acknowledge cancellation before continuing.
         CompactionTasks tasks = cfStore.getCompactionStrategyManager().getMaximalTasks(gcBefore, splitOutput);
 
-        if (tasks.isEmpty())
+        if (tasks == null || tasks.isEmpty())
             return Collections.emptyList();
 
         List<Future<?>> futures = new ArrayList<>();
@@ -998,6 +998,9 @@ public void forceCompactionForTokenRange(ColumnFamilyStore cfStore, Collection<R
                                                                         false,
                                                                         false))
         {
+            if (tasks == null)
+                throw new RuntimeException("Unable to cancel in-progress compactions for " + cfStore.keyspace.getName() + '.' + cfStore.getTableName() + ". Usually retrying will work.");
+
             if (tasks.isEmpty())
                 return;
 

src/java/org/apache/cassandra/db/view/TableViews.java

@@ -89,6 +89,9 @@ public boolean add(View view)
 
     public Iterable<ColumnFamilyStore> allViewsCfs()
     {
+        if (views.isEmpty())
+            return Collections.emptyList();
+
         Keyspace keyspace = Keyspace.open(baseTableMetadata.keyspace);
         return Iterables.transform(views, view -> keyspace.getColumnFamilyStore(view.getDefinition().name()));
     }

src/java/org/apache/cassandra/exceptions/RequestFailureReason.java

@@ -35,7 +35,8 @@ public enum RequestFailureReason
     UNKNOWN                  (0),
     READ_TOO_MANY_TOMBSTONES (1),
     TIMEOUT                  (2),
-    INCOMPATIBLE_SCHEMA      (3);
+    INCOMPATIBLE_SCHEMA      (3),
+    TRUNCATE_FAILED          (12);
 
     public static final Serializer serializer = new Serializer();
 
@@ -85,6 +86,9 @@ public static RequestFailureReason forException(Throwable t)
         if (t instanceof IncompatibleSchemaException)
             return INCOMPATIBLE_SCHEMA;
 
+        if (t instanceof TruncateException)
+            return TRUNCATE_FAILED;
+
         return UNKNOWN;
     }
 

src/java/org/apache/cassandra/net/InboundSink.java

@@ -22,6 +22,8 @@
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Predicate;
 
+import org.apache.cassandra.exceptions.TruncateException;
+
 import org.slf4j.LoggerFactory;
 
 import net.openhft.chronicle.core.util.ThrowingConsumer;
@@ -100,7 +102,9 @@ public void accept(Message<?> message)
         {
             fail(message.header, t);
 
-            if (t instanceof TombstoneOverwhelmingException || t instanceof IndexNotAvailableException)
+            if (t instanceof TombstoneOverwhelmingException ||
+                t instanceof IndexNotAvailableException ||
+                t instanceof TruncateException)
                 noSpamLogger.error(t.getMessage());
             else if (t instanceof RuntimeException)
                 throw (RuntimeException) t;

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

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.db;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.agent.ByteBuddyAgent;
+import net.bytebuddy.dynamic.loading.ClassReloadingStrategy;
+import net.bytebuddy.implementation.StubMethod;
+
+import org.assertj.core.api.Assertions;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
+import org.apache.cassandra.exceptions.TruncateException;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+
+public class TruncateBlockingTest extends CQLTester
+{
+    @BeforeClass
+    public static void setUp()
+    {
+        ByteBuddyAgent.install();
+        // Stub out waitForCessation so the test doesn't wait 60s
+        new ByteBuddy().redefine(CompactionManager.class)
+                .method(named("waitForCessation"))
+                .intercept(StubMethod.INSTANCE)
+                .make()
+                .load(CompactionManager.class.getClassLoader(), ClassReloadingStrategy.fromInstalledAgent());
+    }
+
+    @Test
+    public void testTruncateFailsWhenCompactionsDoNotStopInTime() throws Throwable
+    {
+        createTable("CREATE TABLE %s (id int PRIMARY KEY, v text)");
+
+        execute("INSERT INTO %s (id, v) VALUES (1, 'a')");
+        execute("INSERT INTO %s (id, v) VALUES (2, 'b')");
+        execute("INSERT INTO %s (id, v) VALUES (3, 'c')");
+        flush();
+
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+
+        // Mark the sstable as compacting directly in the tracker, without registering a
+        // CompactionInfo.Holder. There is nothing for interruptCompactionForCFs to stop, so
+        // runWithCompactionsDisabled falls through to waitForCessation, then finds the sstable
+        // still in the compacting set and returns null.
+        try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstable, OperationType.ANTICOMPACTION))
+        {
+            assertNotNull("Unable to mark sstable compacting", txn);
+
+            Assertions.assertThatThrownBy(cfs::truncateBlocking)
+                    .as("Unable to stop compaction. Usually retrying truncate will work")
+                    .isInstanceOf(TruncateException.class);
+
+            assertRows(execute("SELECT * FROM %s WHERE id = 1"), row(1, "a"));
+            assertRows(execute("SELECT * FROM %s WHERE id = 2"), row(2, "b"));
+            assertRows(execute("SELECT * FROM %s WHERE id = 3"), row(3, "c"));
+            assertFalse("SSTables should still be present after truncation failure",
+                    cfs.getLiveSSTables().isEmpty());
+        }
+    }
+
+    @Test
+    public void testRebuildOnFailedScrubReturnsFalseWhenTruncateFails() throws Throwable
+    {
+        createTable("CREATE TABLE %s (id int PRIMARY KEY, v text)");
+        // rebuildOnFailedScrub only applies to indexes with their own backing table
+        createIndex("CREATE INDEX ON %s (v)");
+
+        execute("INSERT INTO %s (id, v) VALUES (1, 'a')");
+        flush();
+
+        ColumnFamilyStore baseCfs = getCurrentColumnFamilyStore();
+        ColumnFamilyStore indexCfs = baseCfs.indexManager.getAllIndexColumnFamilyStores().iterator().next();
+        SSTableReader sstable = indexCfs.getLiveSSTables().iterator().next();
+
+        try (LifecycleTransaction txn = indexCfs.getTracker().tryModify(sstable, OperationType.ANTICOMPACTION))
+        {
+            assertNotNull("Unable to mark sstable compacting", txn);
+
+            RuntimeException scrubFailure = new RuntimeException("original scrub failure");
+            // rebuildOnFailedScrub should report the rebuild as unsuccessful
+            assertFalse("rebuildOnFailedScrub should return false when it can't truncate the index",
+                    indexCfs.rebuildOnFailedScrub(scrubFailure));
+        }
+    }
+}