diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 4ff6e096e7..f9cddc54b7 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/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; - 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 class ColumnFamilyStore implements ColumnFamilyStoreMBean 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,41 +2344,58 @@ 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() - { - 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); - - indexManager.truncateAllIndexesBlocking(truncatedAt); - viewManager.truncateBlocking(replayAfter, truncatedAt); - - SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); - logger.trace("cleaning out row cache"); - invalidateCaches(); - - } - }; - - runWithCompactionsDisabled(Executors.callable(truncateRunnable), true, true); - - viewManager.build(); + 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 + // 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)); + + discardSSTables(truncatedAt); + + indexManager.truncateAllIndexesBlocking(truncatedAt); + viewManager.truncateBlocking(replayAfter, truncatedAt); + + SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); + logger.trace("cleaning out row cache"); + invalidateCaches(); + return true; + } + /** * Drops current memtable without flushing to disk. This should only be called when truncating a column family which is not durable. */ diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index b525435d94..b86fe2e762 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/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> 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; diff --git a/src/java/org/apache/cassandra/db/view/TableViews.java b/src/java/org/apache/cassandra/db/view/TableViews.java index 8e64ef3589..3725603e43 100644 --- a/src/java/org/apache/cassandra/db/view/TableViews.java +++ b/src/java/org/apache/cassandra/db/view/TableViews.java @@ -89,6 +89,9 @@ public class TableViews extends AbstractCollection public Iterable allViewsCfs() { + if (views.isEmpty()) + return Collections.emptyList(); + Keyspace keyspace = Keyspace.open(baseTableMetadata.keyspace); return Iterables.transform(views, view -> keyspace.getColumnFamilyStore(view.getDefinition().name())); } diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java index 1cdbdb544d..9d9bf74d96 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java +++ b/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; } diff --git a/src/java/org/apache/cassandra/net/InboundSink.java b/src/java/org/apache/cassandra/net/InboundSink.java index 16eb440540..ead0aa5672 100644 --- a/src/java/org/apache/cassandra/net/InboundSink.java +++ b/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; diff --git a/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java b/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java new file mode 100644 index 0000000000..fcceee603b --- /dev/null +++ b/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)); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java index beed019554..7ea27a1367 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java +++ b/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 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 sstables = createSSTables(cfs, 3, 0); + + LifecycleTransaction txn = blockCompactions(cfs, sstables); + ClassReloadingStrategy strategy = stubWaitForCessation(); + try + { + Range 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 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 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 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 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 createSSTables(ColumnFamilyStore cfs, int count, int startGeneration) { List sstables = new ArrayList<>();