This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-4.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-4.0 by this push:
new 7b4170491d Make runWithCompactionsDisabled return non-null on success
7b4170491d is described below
commit 7b4170491d124cd817278e01f0ff3f09012db889
Author: nivy <[email protected]>
AuthorDate: Thu Jul 30 17:02:26 2026 -0700
Make runWithCompactionsDisabled return non-null on success
patch by Nivy Kani; reviewed by Stefan Miklosovic and Caleb Rackliffe
---
CHANGES.txt | 1 +
.../org/apache/cassandra/db/ColumnFamilyStore.java | 89 ++++++++++-----
.../cassandra/db/compaction/CompactionManager.java | 5 +-
.../org/apache/cassandra/db/view/TableViews.java | 3 +
.../cassandra/exceptions/RequestFailureReason.java | 8 +-
src/java/org/apache/cassandra/net/InboundSink.java | 5 +-
.../apache/cassandra/db/TruncateBlockingTest.java | 99 +++++++++++++++++
.../db/compaction/CancelCompactionsTest.java | 121 +++++++++++++++++++++
8 files changed, 300 insertions(+), 31 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index dbcce9d96f..50ad389065 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0.22
+ * Make runWithCompactionsDisabled return non-null on success (CASSANDRA-21527)
* Validate authz before performing role check in LIST ROLES/PERMISSIONS
(CASSANDRA-21560)
* Ensure transferred_ranges reset on decommision re-attempt when pending
ranges cannot be proven continous (CASSANDRA-16290)
* Add validation to uncompressed length during decompression (CASSANDRA-21567)
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 4ff6e096e7..2ac4783d89 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,39 +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 in-progress
compactions. Please retry when current compaction tasks have completed or
overall system load has decreased.");
+ else if (!succeeded)
+ failure = new TruncateException("Truncate failed.");
+ }
+ 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);
+ maybeFail(failure);
- indexManager.truncateAllIndexesBlocking(truncatedAt);
- viewManager.truncateBlocking(replayAfter, truncatedAt);
+ logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+ }
- SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this,
truncatedAt, replayAfter);
- logger.trace("cleaning out row cache");
- invalidateCaches();
+ 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));
- runWithCompactionsDisabled(Executors.callable(truncateRunnable), true,
true);
+ discardSSTables(truncatedAt);
- viewManager.build();
+ indexManager.truncateAllIndexesBlocking(truncatedAt);
+ viewManager.truncateBlocking(replayAfter, truncatedAt);
- logger.info("Truncate of {}.{} is complete", keyspace.getName(), name);
+ SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this,
truncatedAt, replayAfter);
+ logger.trace("cleaning out row cache");
+ invalidateCaches();
+ return true;
}
/**
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<Future<?>> 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<View>
public Iterable<ColumnFamilyStore> 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..6664070e97 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();
@@ -74,7 +75,7 @@ public enum RequestFailureReason
throw new IllegalArgumentException("RequestFailureReason code must
be non-negative (got " + code + ')');
// be forgiving and return UNKNOWN if we aren't aware of the code -
for forward compatibility
- return code < codeToReasonMap.length ? codeToReasonMap[code] : UNKNOWN;
+ return code < codeToReasonMap.length && codeToReasonMap[code] != null
? codeToReasonMap[code] : UNKNOWN;
}
public static RequestFailureReason forException(Throwable t)
@@ -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..2cc5d24b4c 100644
--- a/src/java/org/apache/cassandra/net/InboundSink.java
+++ b/src/java/org/apache/cassandra/net/InboundSink.java
@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
import net.openhft.chronicle.core.util.ThrowingConsumer;
import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
import org.apache.cassandra.exceptions.RequestFailureReason;
+import org.apache.cassandra.exceptions.TruncateException;
import org.apache.cassandra.index.IndexNotAvailableException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.NoSpamLogger;
@@ -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/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java
b/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java
new file mode 100644
index 0000000000..24433a6306
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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 org.assertj.core.api.Assertions;
+import org.jboss.byteman.contrib.bmunit.BMRule;
+import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.cassandra.cql3.CQLTester;
+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 org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+
+@RunWith(BMUnitRunner.class)
+public class TruncateBlockingTest extends CQLTester
+{
+ @Test
+ @BMRule(name = "no-op waitForCessation",
+ targetClass =
"org.apache.cassandra.db.compaction.CompactionManager",
+ targetMethod = "waitForCessation",
+ action = "return;")
+ 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<SSTableReader> 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<SSTableReader> sstables = createSSTables(cfs, 3, 0);
+
+ LifecycleTransaction txn = blockCompactions(cfs, sstables);
+ ClassReloadingStrategy strategy = stubWaitForCessation();
+ 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
+ {
+ strategy.reset(CompactionManager.class);
+ txn.abort();
+ }
+ }
+
+ @Test
+ public void
testGarbageCollectReturnsUnableToCancelWhenCompactionsCannotBeDisabled() throws
Throwable
+ {
+ ColumnFamilyStore cfs = MockSchema.newCFS();
+ List<SSTableReader> 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<SSTableReader> 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<UUID> 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<SSTableReader> 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<SSTableReader> createSSTables(ColumnFamilyStore cfs, int
count, int startGeneration)
{
List<SSTableReader> sstables = new ArrayList<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
