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 

Fix flaky test to wait for the index

Formatting

Fix test

Fix test
This commit is contained in:
nivy 2026-07-16 16:37:56 -07:00
parent 1515da5635
commit c2e6d75bb4
9 changed files with 359 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

5
src/java/org/apache/cassandra/exceptions/RequestFailure.java
View File

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

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

@ -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())
{

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

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

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

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

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

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

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

1
test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java
View File

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