Merge b40cb8e73c into 172099f4a8

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

@@ -90,6 +90,7 @@ import org.apache.cassandra.utils.concurrent.Refs;
 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 class ColumnFamilyStore implements ColumnFamilyStoreMBean
         if (!isIndex() || !SecondaryIndexManager.isIndexColumnFamilyStore(this))
             return false;
 
+        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 class ColumnFamilyStore implements ColumnFamilyStoreMBean
                 UUID session = sst.getPendingRepair();
                 return session != null && sessions.contains(session);
             };
-            return runWithCompactionsDisabled(() -> compactionStrategyManager.releaseRepairData(sessions),
+            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,10 +2344,35 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             for (SSTableReader sstable : cfs.getLiveSSTables())
                 now = Math.max(now, sstable.maxDataAge);
         truncatedAt = now;
-
-        Runnable truncateRunnable = new Runnable()
+        Throwable failure = null;
+        try
         {
-            public void run()
+            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;
+        }
+
+        try
+        {
+            viewManager.build();
+        }
+        catch (Throwable t)
+        {
+            failure = merge(failure, t);
+        }
+
+        maybeFail(failure);
+
+        logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+    }
+
+    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
@@ -2352,15 +2393,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter);
         logger.trace("cleaning out row cache");
         invalidateCaches();
-
-            }
-        };
-
-        runWithCompactionsDisabled(Executors.callable(truncateRunnable), true, true);
-
-        viewManager.build();
-
-        logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+        return true;
     }
 
     /**

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

@@ -942,7 +942,7 @@ public class CompactionManager implements CompactionManagerMBean
         // 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 class CompactionManager implements CompactionManagerMBean
                                                                         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 class TableViews extends AbstractCollection<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 enum RequestFailureReason
         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.TimeUnit;
 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 class InboundSink implements InboundMessageHandlers.MessageConsumer
         {
             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));
+        }
+    }
+}

test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java

@@ -33,6 +33,11 @@ import java.util.stream.Collectors;
 
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.Uninterruptibles;
+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.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -64,6 +69,10 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
+import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
 
 public class CancelCompactionsTest extends CQLTester
 {
@@ -363,6 +372,118 @@ public class CancelCompactionsTest extends CQLTester
         return new Murmur3Partitioner.LongToken(t);
     }
 
+    private LifecycleTransaction blockCompactions(ColumnFamilyStore cfs, List<SSTableReader> sstables)
+    {
+        LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.COMPACTION);
+        assertNotNull(txn);
+        return txn;
+    }
+
+    private ClassReloadingStrategy stubWaitForCessation()
+    {
+        ByteBuddyAgent.install();
+        ClassReloadingStrategy strategy = ClassReloadingStrategy.fromInstalledAgent();
+        new ByteBuddy().redefine(CompactionManager.class)
+                       .method(named("waitForCessation"))
+                       .intercept(StubMethod.INSTANCE)
+                       .make()
+                       .load(CompactionManager.class.getClassLoader(), strategy);
+        return strategy;
+    }
+
+    @Test
+    public void testForceCompactionThrowsWhenCompactionsCannotBeDisabled() throws Exception
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS();
+        List<SSTableReader> sstables = createSSTables(cfs, 3, 0);
+
+        LifecycleTransaction txn = blockCompactions(cfs, sstables);
+        ClassReloadingStrategy strategy = stubWaitForCessation();
+        try
+        {
+            Range<Token> allData = new Range<>(cfs.getPartitioner().getMinimumToken(), cfs.getPartitioner().getMaximumToken());
+
+            // forceCompaction should fail at the null runWithCompactionsDisabled result
+            Assertions.assertThatThrownBy(() -> cfs.forceCompactionForTokenRange(Collections.singleton(allData)))
+                      .as("Unable to cancel in-progress compactions. Usually retrying will work")
+                      .isInstanceOf(RuntimeException.class);
+        }
+        finally
+        {
+            strategy.reset(CompactionManager.class);
+            txn.abort();
+        }
+    }
+
+    @Test
+    public void testGarbageCollectReturnsUnableToCancelWhenCompactionsCannotBeDisabled() throws Throwable
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS();
+        List<SSTableReader> sstables = createSSTables(cfs, 3, 0);
+
+        LifecycleTransaction txn = blockCompactions(cfs, sstables);
+        ClassReloadingStrategy strategy = stubWaitForCessation();
+        try
+        {
+            // garbageCollect goes through parallelAllSSTableOperation, which must handle a null
+            // LifecycleTransaction from markAllCompacting without NPEing.
+            assertEquals(CompactionManager.AllSSTableOpStatus.UNABLE_TO_CANCEL, cfs.garbageCollect(TombstoneOption.ROW, 0));
+        }
+        finally
+        {
+            strategy.reset(CompactionManager.class);
+            txn.abort();
+        }
+    }
+
+    @Test
+    public void testReleaseRepairDataReturnsUnsuccessfulWhenCompactionsCannotBeDisabled() throws Exception
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS();
+        List<SSTableReader> sstables = createSSTables(cfs, 3, 0);
+
+        // releaseRepairData only tries to cancel compactions on sstables pending one of these sessions,
+        // so tag one of our blocked sstables with a pending session to make it match
+        UUID pendingSession = UUID.randomUUID();
+        Set<UUID> sessions = ImmutableSet.of(pendingSession, UUID.randomUUID());
+        AbstractPendingRepairTest.mutateRepaired(sstables.get(0), pendingSession, false);
+
+        LifecycleTransaction txn = blockCompactions(cfs, sstables);
+        ClassReloadingStrategy strategy = stubWaitForCessation();
+        try
+        {
+            // force release should report the sessions as unsuccessful rather than throwing
+            CleanupSummary summary = cfs.releaseRepairData(sessions, true);
+            assertTrue(summary.successful.isEmpty());
+            assertEquals(sessions, summary.unsuccessful);
+        }
+        finally
+        {
+            strategy.reset(CompactionManager.class);
+            txn.abort();
+        }
+    }
+
+    @Test
+    public void testSubmitMaximalNoOpsWhenCompactionsCannotBeDisabled() throws Exception
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS();
+        List<SSTableReader> sstables = createSSTables(cfs, 3, 0);
+
+        LifecycleTransaction txn = blockCompactions(cfs, sstables);
+        ClassReloadingStrategy strategy = stubWaitForCessation();
+        try
+        {
+            // submitMaximal should no-op on null getMaximalTasks result
+            assertTrue(CompactionManager.instance.submitMaximal(cfs, -1, false).isEmpty());
+        }
+        finally
+        {
+            strategy.reset(CompactionManager.class);
+            txn.abort();
+        }
+    }
+
     private List<SSTableReader> createSSTables(ColumnFamilyStore cfs, int count, int startGeneration)
     {
         List<SSTableReader> sstables = new ArrayList<>();