From b58bec1edef753d52f3d44ecf5d70aefb09efcc6 Mon Sep 17 00:00:00 2001 From: nivy Date: Thu, 30 Jul 2026 15:49:30 -0700 Subject: [PATCH] Make runWithCompactionsDisabled return non-null on success --- .../cassandra/db/ColumnFamilyStore.java | 94 ++++--- .../db/compaction/CompactionManager.java | 5 +- .../exceptions/RequestFailureReason.java | 6 +- .../org/apache/cassandra/net/InboundSink.java | 5 +- .../cassandra/service/StorageService.java | 2 + .../cassandra/db/TruncateBlockingTest.java | 237 ++++++++++++++++++ .../db/compaction/CancelCompactionsTest.java | 112 +++++++++ 7 files changed, 427 insertions(+), 34 deletions(-) create mode 100644 test/unit/org/apache/cassandra/db/TruncateBlockingTest.java diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index fd47b8e5bd..d8900e2bb5 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -114,6 +114,7 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.exceptions.StartupException; +import org.apache.cassandra.exceptions.TruncateException; import org.apache.cassandra.index.SecondaryIndexManager; import org.apache.cassandra.index.internal.CassandraIndex; import org.apache.cassandra.index.transactions.UpdateTransaction; @@ -1782,7 +1783,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner 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()); @@ -1904,8 +1913,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner TimeUUID session = sst.getPendingRepair(); return session != null && sessions.contains(session); }; - return runWithCompactionsDisabled(() -> compactionStrategyManager.releaseRepairData(sessions), - predicate, OperationType.STREAM, false, true, true); + CleanupSummary summary = runWithCompactionsDisabled(() -> compactionStrategyManager.releaseRepairData(sessions), + predicate, OperationType.STREAM, false, true, true); + if (summary == null) + { + logger.warn("Unable to cancel in-progress compactions for {}.{}, could not force release repair data for sessions {}", + getKeyspaceName(), name, sessions); + return new CleanupSummary(this, Collections.emptySet(), new HashSet<>(sessions)); + } + return summary; } else { @@ -2794,41 +2810,57 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner 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={}", getKeyspaceName(), 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), OperationType.P0, 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 && isAutoSnapshotEnabled()) - snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(name, SNAPSHOT_TRUNCATE_PREFIX), DatabaseDescriptor.getAutoSnapshotTtl()); + 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(FutureTask.callable(truncateRunnable), OperationType.P0, true, true); - - viewManager.build(); + maybeFail(failure); logger.info("Truncate of {}.{} is complete", getKeyspaceName(), name); } + private boolean runTruncate(long truncatedAt, boolean noSnapshot, CommitLogPosition replayAfter) + { + logger.info("Truncating {}.{} with truncatedAt={}", getKeyspaceName(), 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 && isAutoSnapshotEnabled()) + snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(name, SNAPSHOT_TRUNCATE_PREFIX), DatabaseDescriptor.getAutoSnapshotTtl()); + + 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 * that cannot have dirty intervals in the commit log (i.e. one which is not durable, or where the memtable itself diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 7c93558004..a20dc32a6e 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -1000,7 +1000,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan // for ourselves to finish/acknowledge cancellation before continuing. CompactionTasks tasks = cfStore.getCompactionStrategyManager().getMaximalTasks(gcBefore, splitOutput, operationType); - if (tasks.isEmpty()) + if (tasks == null || tasks.isEmpty()) return Collections.emptyList(); List> futures = new ArrayList<>(); @@ -1048,6 +1048,9 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan false, false)) { + if (tasks == null) + throw new RuntimeException("Unable to cancel in-progress compactions for " + cfStore.getKeyspaceName() + '.' + cfStore.getTableName() + ". Usually retrying will work."); + if (tasks.isEmpty()) return; diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java index 5e9900280c..1a37dabbb7 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java @@ -37,7 +37,8 @@ public enum RequestFailureReason READ_SIZE (4), NODE_DOWN (5), INDEX_NOT_AVAILABLE (6), - READ_TOO_MANY_INDEXES (7); + READ_TOO_MANY_INDEXES (7), + TRUNCATE_FAILED (12); public static final Serializer serializer = new Serializer(); @@ -87,6 +88,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 9d68ba1aa0..78b105be1e 100644 --- a/src/java/org/apache/cassandra/net/InboundSink.java +++ b/src/java/org/apache/cassandra/net/InboundSink.java @@ -30,6 +30,7 @@ import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.index.IndexNotAvailableException; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.exceptions.TruncateException; /** * A message sink that all inbound messages go through. @@ -100,7 +101,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/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 22ef542f2a..edad65f927 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -7762,6 +7762,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE table.getCompactionStrategyManager().mutateRepaired(sstables, repairedAt, null, false); return sstables; }, predicate, OperationType.ANTICOMPACTION, true, false, true); + if (result == null) + throw new RuntimeException("Unable to cancel in-progress compactions for " + keyspace + '.' + tableName + ". Usually retrying will work."); sstablesTouched.addAll(result.stream().map(sst -> sst.descriptor.baseFile().name()).collect(Collectors.toList())); } return sstablesTouched; 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..3da4ef2086 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java @@ -0,0 +1,237 @@ +/* + * 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 java.util.Collections; + +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.CompactionInfo; +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 org.apache.cassandra.service.StorageService; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; +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 testTruncateFailsWhenCompactionsCannotBeDisabled() + { + 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(); + + // Register a P0-priority compaction holder for this table to force + // runWithCompactionsDisabled to return null immediately. + CompactionInfo.Holder holder = new CompactionInfo.Holder() + { + public CompactionInfo getCompactionInfo() + { + return new CompactionInfo(cfs.metadata(), + OperationType.P0, + 0, + 100, + 100, + nextTimeUUID(), + Collections.emptySet()); + } + + public boolean isGlobal() + { + return false; + } + }; + + CompactionManager.instance.active.beginCompaction(holder); + try + { + 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()); + + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + public void testTruncateFailsWhenCompactionsDoNotStopInTime() + { + 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() + { + 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) USING 'legacy_local_table'"); + + execute("INSERT INTO %s (id, v) VALUES (1, 'a')"); + flush(); + + ColumnFamilyStore baseCfs = getCurrentColumnFamilyStore(); + ColumnFamilyStore indexCfs = baseCfs.indexManager.getAllIndexColumnFamilyStores().iterator().next(); + + // Register a P0-priority compaction holder for the index cfs to force truncateBlocking to fail. + CompactionInfo.Holder holder = new CompactionInfo.Holder() + { + public CompactionInfo getCompactionInfo() + { + return new CompactionInfo(indexCfs.metadata(), + OperationType.P0, + 0, + 100, + 100, + nextTimeUUID(), + Collections.emptySet()); + } + + public boolean isGlobal() + { + return false; + } + }; + + CompactionManager.instance.active.beginCompaction(holder); + try + { + 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)); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + public void testMutateSSTableRepairedStateThrowsWhenCompactionsCannotBeDisabled() + { + createTable("CREATE TABLE %s (id int PRIMARY KEY, v text)"); + + execute("INSERT INTO %s (id, v) VALUES (1, 'a')"); + flush(); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + + // Register a P0-priority compaction holder for this table to force runWithCompactionsDisabled + // to return null + CompactionInfo.Holder holder = new CompactionInfo.Holder() + { + public CompactionInfo getCompactionInfo() + { + return new CompactionInfo(cfs.metadata(), + OperationType.P0, + 0, + 100, + 100, + nextTimeUUID(), + Collections.emptySet()); + } + + public boolean isGlobal() + { + return false; + } + }; + + CompactionManager.instance.active.beginCompaction(holder); + try + { + // mutateSSTableRepairedState should report the null runWithCompactionsDisabled result as a + // failure to the caller + Assertions.assertThatThrownBy(() -> StorageService.instance.mutateSSTableRepairedState(true, false, keyspace(), Collections.singletonList(currentTable()))) + .as("Unable to cancel in-progress compactions. Usually retrying will work") + .isInstanceOf(RuntimeException.class); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java index b3a182f186..88a3987b33 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java @@ -32,6 +32,7 @@ import java.util.stream.Collectors; import com.google.common.collect.ImmutableSet; import com.google.common.util.concurrent.Uninterruptibles; +import org.assertj.core.api.Assertions; import org.junit.Assume; import org.junit.Test; @@ -54,6 +55,8 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.RangesAtEndpoint; import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.repair.consistent.admin.CleanupSummary; +import org.apache.cassandra.schema.CompactionParams.TombstoneOption; import org.apache.cassandra.schema.MockSchema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; @@ -372,6 +375,115 @@ public class CancelCompactionsTest extends CQLTester return new Murmur3Partitioner.LongToken(t); } + private CompactionInfo.Holder p0Holder(ColumnFamilyStore cfs) + { + return new CompactionInfo.Holder() + { + public CompactionInfo getCompactionInfo() + { + return new CompactionInfo(cfs.metadata(), OperationType.P0, 0, 100, 100, nextTimeUUID(), Collections.emptySet()); + } + + public boolean isGlobal() + { + return false; + } + }; + } + + @Test + public void testForceCompactionThrowsWhenCompactionsCannotBeDisabled() + { + ColumnFamilyStore cfs = MockSchema.newCFS(); + createSSTables(cfs, 3, 0); + + // Register a P0-priority compaction holder for this table to force runWithCompactionsDisabled + // to return null + CompactionInfo.Holder holder = p0Holder(cfs); + CompactionManager.instance.active.beginCompaction(holder); + 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 + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + public void testGarbageCollectReturnsUnableToCancelWhenCompactionsCannotBeDisabled() throws Throwable + { + ColumnFamilyStore cfs = MockSchema.newCFS(); + createSSTables(cfs, 3, 0); + + // Register a P0-priority compaction holder for this table to force runWithCompactionsDisabled + // to return null immediately. + CompactionInfo.Holder holder = p0Holder(cfs); + CompactionManager.instance.active.beginCompaction(holder); + try + { + // garbageCollect goes through withAllSSTables, which must be able to pass a null + // LifecycleTransaction to its caller-supplied op without NPEing. + assertEquals(CompactionManager.AllSSTableOpStatus.UNABLE_TO_CANCEL, cfs.garbageCollect(TombstoneOption.ROW, 0)); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + public void testReleaseRepairDataReturnsUnsuccessfulWhenCompactionsCannotBeDisabled() + { + ColumnFamilyStore cfs = MockSchema.newCFS(); + createSSTables(cfs, 3, 0); + + Set sessions = ImmutableSet.of(nextTimeUUID(), nextTimeUUID()); + + // Register a P0-priority compaction holder for this table to force runWithCompactionsDisabled + // to return null immediately. + CompactionInfo.Holder holder = p0Holder(cfs); + CompactionManager.instance.active.beginCompaction(holder); + 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 + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + public void testSubmitMaximalNoOpsWhenCompactionsCannotBeDisabled() + { + ColumnFamilyStore cfs = MockSchema.newCFS(); + createSSTables(cfs, 3, 0); + + // Register a P0-priority compaction holder for this table to force runWithCompactionsDisabled + // to return null immediately. + CompactionInfo.Holder holder = p0Holder(cfs); + CompactionManager.instance.active.beginCompaction(holder); + try + { + // submitMaximal should no-op on null getMaximalTasks result + assertTrue(CompactionManager.instance.submitMaximal(cfs, -1, false).isEmpty()); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + private List createSSTables(ColumnFamilyStore cfs, int count, int startGeneration) { List sstables = new ArrayList<>();