natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make runWithCompactionsDisabled return non-null on success

This commit is contained in:
nivy 2026-07-30 15:49:30 -07:00
parent ecc0a3e77b
commit b58bec1ede
7 changed files with 427 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -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

5
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -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<Future<?>> 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;

6
src/java/org/apache/cassandra/exceptions/RequestFailureReason.java
View File

@ -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;
}

5
src/java/org/apache/cassandra/net/InboundSink.java
View File

@ -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;

2
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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;

237
test/unit/org/apache/cassandra/db/TruncateBlockingTest.java Normal file
View File

@ -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);
}
}
}

112
test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java
View File

@ -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<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
{
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<TimeUUID> 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<SSTableReader> createSSTables(ColumnFamilyStore cfs, int count, int startGeneration)
{
List<SSTableReader> sstables = new ArrayList<>();