From c2e6d75bb4d64b86d25283f706fad9abfe1bb85f Mon Sep 17 00:00:00 2001 From: nivy Date: Thu, 16 Jul 2026 16:37:56 -0700 Subject: [PATCH] Make runWithCompactionsDisabled return non-null on success Fix flaky test to wait for the index Formatting Fix test Fix test --- .../cassandra/db/ColumnFamilyStore.java | 67 +++-- .../db/compaction/CompactionManager.java | 3 + .../cassandra/exceptions/RequestFailure.java | 5 + .../exceptions/RequestFailureReason.java | 2 + .../org/apache/cassandra/net/InboundSink.java | 4 +- .../cassandra/service/StorageService.java | 2 + .../cassandra/db/TruncateBlockingTest.java | 237 ++++++++++++++++++ .../db/compaction/CancelCompactionsTest.java | 65 +++++ .../exceptions/RequestFailureReasonTest.java | 1 + 9 files changed, 359 insertions(+), 27 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 db15db6729..f7dd671aae 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -112,6 +112,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; @@ -1830,7 +1831,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()); @@ -2627,38 +2636,44 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner for (SSTableReader sstable : cfs.getLiveSSTables()) now = Math.max(now, sstable.maxDataAge); truncatedAt = now; - - Runnable truncateRunnable = new Runnable() + try { - public void run() + 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) { - 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(noSnapshot, truncatedAt, DatabaseDescriptor.getAutoSnapshotTtl()); - - discardSSTables(truncatedAt); - - indexManager.truncateAllIndexesBlocking(truncatedAt); - viewManager.truncateBlocking(replayAfter, truncatedAt); - - SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); - logger.trace("cleaning out row cache"); - invalidateCaches(); - + throw new TruncateException("Unable to stop compaction. Usually retrying truncate will work."); } - }; - - runWithCompactionsDisabled(FutureTask.callable(truncateRunnable), OperationType.P0, true, true); - - viewManager.build(); + } + finally + { + viewManager.build(); + } 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(noSnapshot, truncatedAt, 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 52866c045c..00626dd6dc 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -1248,6 +1248,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/RequestFailure.java b/src/java/org/apache/cassandra/exceptions/RequestFailure.java index 4990c25973..dc9e06a8c0 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailure.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailure.java @@ -56,6 +56,7 @@ public class RequestFailure public static final RequestFailure READ_TOO_MANY_INDEXES = new RequestFailure(RequestFailureReason.READ_TOO_MANY_INDEXES); public static final RequestFailure RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM = new RequestFailure(RequestFailureReason.RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM); public static final RequestFailure INDEX_BUILD_IN_PROGRESS = new RequestFailure(RequestFailureReason.INDEX_BUILD_IN_PROGRESS); + public static final RequestFailure TRUNCATE_FAILED = new RequestFailure(RequestFailureReason.TRUNCATE_FAILED); static { @@ -126,6 +127,9 @@ public class RequestFailure if (t instanceof CoordinatorBehindException) return COORDINATOR_BEHIND; + if (t instanceof TruncateException) + return TRUNCATE_FAILED; + return new RequestFailure(t); } @@ -147,6 +151,7 @@ public class RequestFailure case READ_TOO_MANY_INDEXES: return READ_TOO_MANY_INDEXES; case INDEX_BUILD_IN_PROGRESS: return INDEX_BUILD_IN_PROGRESS; case RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM: return RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM; + case TRUNCATE_FAILED: return TRUNCATE_FAILED; } } diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java index 4e074c9173..fbaa7b7da5 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java @@ -51,6 +51,7 @@ public enum RequestFailureReason INVALID_ROUTING (9), COORDINATOR_BEHIND (10), RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM (11), + TRUNCATE_FAILED (12), // The following codes have been ported from an external fork, where they were offset explicitly to avoid conflicts. INDEX_BUILD_IN_PROGRESS (503), ; @@ -95,6 +96,7 @@ public enum RequestFailureReason exceptionToReasonMap.put(CoordinatorBehindException.class, COORDINATOR_BEHIND); exceptionToReasonMap.put(IndexBuildInProgressException.class, INDEX_BUILD_IN_PROGRESS); exceptionToReasonMap.put(RetryOnDifferentSystemException.class, RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM); + exceptionToReasonMap.put(TruncateException.class, TRUNCATE_FAILED); if (exceptionToReasonMap.size() != reasons.length - withoutExceptions.size()) { diff --git a/src/java/org/apache/cassandra/net/InboundSink.java b/src/java/org/apache/cassandra/net/InboundSink.java index 734c9a1ba1..6de23ede46 100644 --- a/src/java/org/apache/cassandra/net/InboundSink.java +++ b/src/java/org/apache/cassandra/net/InboundSink.java @@ -31,6 +31,7 @@ import org.apache.cassandra.db.filter.TombstoneOverwhelmingException; import org.apache.cassandra.exceptions.CoordinatorBehindException; import org.apache.cassandra.exceptions.InvalidRoutingException; import org.apache.cassandra.exceptions.RequestFailure; +import org.apache.cassandra.exceptions.TruncateException; import org.apache.cassandra.index.IndexBuildInProgressException; import org.apache.cassandra.index.IndexNotAvailableException; import org.apache.cassandra.locator.InetAddressAndPort; @@ -134,7 +135,8 @@ public class InboundSink implements InboundMessageHandlers.MessageConsumer else if (t instanceof TombstoneOverwhelmingException || t instanceof IndexNotAvailableException || t instanceof IndexBuildInProgressException || - t instanceof InvalidRoutingException) + t instanceof InvalidRoutingException || + t instanceof TruncateException) { noSpamLogger.error(t.getMessage()); } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 397e50fa0d..0b7900f0bd 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -5890,6 +5890,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..c83b1ac4a7 --- /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); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java index f707a16763..37cd18e1a5 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java @@ -34,6 +34,7 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import com.google.common.util.concurrent.Uninterruptibles; +import org.assertj.core.api.Assertions; import org.junit.Assume; import org.junit.Test; @@ -56,6 +57,7 @@ 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.schema.CompactionParams.TombstoneOption; import org.apache.cassandra.schema.MockSchema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; @@ -374,6 +376,69 @@ 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().getMaximumTokenForSplitting()); + + // 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); + } + } + private List createSSTables(ColumnFamilyStore cfs, int count, int startGeneration) { List sstables = new ArrayList<>(); diff --git a/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java b/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java index 4be82d491c..35f186ad4e 100644 --- a/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java +++ b/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java @@ -41,6 +41,7 @@ public class RequestFailureReasonTest { 9, "INVALID_ROUTING" }, { 10, "COORDINATOR_BEHIND" }, { 11, "RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM" }, + { 12, "TRUNCATE_FAILED" }, { 503, "INDEX_BUILD_IN_PROGRESS" }, };