This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-6.0 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 60b833a8f442de2165b5ae380594015b836d1fef Merge: 7085aa5242 3944e97482 Author: Caleb Rackliffe <[email protected]> AuthorDate: Mon Aug 17 15:01:12 2026 -0500 Merge branch 'cassandra-5.0' into cassandra-6.0 * cassandra-5.0: Make runWithCompactionsDisabled return non-null on success CHANGES.txt | 1 + .../org/apache/cassandra/db/ColumnFamilyStore.java | 84 +++++--- .../cassandra/db/compaction/CompactionManager.java | 5 +- .../cassandra/exceptions/RequestFailure.java | 5 + .../cassandra/exceptions/RequestFailureReason.java | 2 + src/java/org/apache/cassandra/net/InboundSink.java | 4 +- .../apache/cassandra/service/StorageService.java | 2 + .../apache/cassandra/db/TruncateBlockingTest.java | 226 +++++++++++++++++++++ .../db/compaction/CancelCompactionsTest.java | 112 ++++++++++ .../exceptions/RequestFailureReasonTest.java | 1 + 10 files changed, 415 insertions(+), 27 deletions(-) diff --cc CHANGES.txt index 1dd6a5db8f,6e9059397f..dfa56b6881 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -107,8 -26,7 +107,9 @@@ Merged from 5.0 * Ensure SAI sends range tombstones to the coordinator for queries on static columns (CASSANDRA-21332) Merged from 4.1: * Add Paxos v2 option and informatin in cassandra.yaml (CASSANDRA-21316) + * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290) Merged from 4.0: ++ * Make runWithCompactionsDisabled return non-null on success (CASSANDRA-21527) * Bound declared value length against readable bytes in CBUtil (CASSANDRA-21521) * Verify extension type before initializing reflectively-loaded classes (CASSANDRA-21525) * Rename conflicting nodetool import --copy-data short option from -p to -cd (CASSANDRA-20214) diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java index c691d96834,0ab15db8b3..50e962ac04 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@@ -2622,36 -2810,57 +2638,54 @@@ public class ColumnFamilyStore implemen 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(noSnapshot, truncatedAt, DatabaseDescriptor.getAutoSnapshotTtl()); + 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 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; + } - discardSSTables(truncatedAt); + try + { + viewManager.build(); + } + catch (Throwable t) + { + failure = merge(failure, t); + } - indexManager.truncateAllIndexesBlocking(truncatedAt); - viewManager.truncateBlocking(replayAfter, truncatedAt); + maybeFail(failure); - SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); - logger.trace("cleaning out row cache"); - invalidateCaches(); + 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()); ++ data.notifyTruncated(noSnapshot, truncatedAt, DatabaseDescriptor.getAutoSnapshotTtl()); - runWithCompactionsDisabled(FutureTask.callable(truncateRunnable), OperationType.P0, true, true); + discardSSTables(truncatedAt); - viewManager.build(); + indexManager.truncateAllIndexesBlocking(truncatedAt); + viewManager.truncateBlocking(replayAfter, truncatedAt); - logger.info("Truncate of {}.{} is complete", getKeyspaceName(), name); + SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); + logger.trace("cleaning out row cache"); + invalidateCaches(); + return true; } /** diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java index e7ae548ffd,a20dc32a6e..f641a606b0 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@@ -1175,9 -998,9 +1175,9 @@@ public class CompactionManager implemen // here we compute the task off the compaction executor, so having that present doesn't // confuse runWithCompactionsDisabled -- i.e., we don't want to deadlock ourselves, waiting // for ourselves to finish/acknowledge cancellation before continuing. - CompactionTasks tasks = cfStore.getCompactionStrategyManager().getMaximalTasks(gcBefore, splitOutput, operationType); + CompactionTasks tasks = cfStore.getCompactionStrategyManager().getMaximalTasks(gcBefore, splitOutput, permittedParallelism, operationType); - if (tasks.isEmpty()) + if (tasks == null || tasks.isEmpty()) return Collections.emptyList(); List<Future<?>> futures = new ArrayList<>(); diff --cc src/java/org/apache/cassandra/exceptions/RequestFailure.java index 4990c25973,0000000000..dc9e06a8c0 mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailure.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailure.java @@@ -1,178 -1,0 +1,183 @@@ +/* + * 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.exceptions; + +import java.io.IOException; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import org.apache.cassandra.db.filter.TombstoneOverwhelmingException; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.tcm.NotCMSException; + +import static com.google.common.base.Preconditions.checkNotNull; +import static org.apache.cassandra.exceptions.ExceptionSerializer.nullableRemoteExceptionSerializer; + +/** + * Allow inclusion of a serialized exception in failure response messages + * This continues to use the same verb as the old failure response (whether a message payload or parameter) + * and has a nullable failure field that may contain a serialized exception in later versions. + * + * It's important to note RequestFailure is not a singleton for each type, unlike RequestFailureReason, + * since it might include a stack trace so don't compare using identity. + */ +public class RequestFailure +{ + public static final RequestFailure UNKNOWN = new RequestFailure(RequestFailureReason.UNKNOWN); + public static final RequestFailure READ_TOO_MANY_TOMBSTONES = new RequestFailure(RequestFailureReason.READ_TOO_MANY_TOMBSTONES); + public static final RequestFailure TIMEOUT = new RequestFailure(RequestFailureReason.TIMEOUT); + public static final RequestFailure INCOMPATIBLE_SCHEMA = new RequestFailure(RequestFailureReason.INCOMPATIBLE_SCHEMA); + public static final RequestFailure READ_SIZE = new RequestFailure(RequestFailureReason.READ_SIZE); + public static final RequestFailure NODE_DOWN = new RequestFailure(RequestFailureReason.NODE_DOWN); + public static final RequestFailure NOT_CMS = new RequestFailure(RequestFailureReason.NOT_CMS); + public static final RequestFailure INVALID_ROUTING = new RequestFailure(RequestFailureReason.INVALID_ROUTING); + public static final RequestFailure INDEX_NOT_AVAILABLE = new RequestFailure(RequestFailureReason.INDEX_NOT_AVAILABLE); + public static final RequestFailure COORDINATOR_BEHIND = new RequestFailure(RequestFailureReason.COORDINATOR_BEHIND); + 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 + { + // Validate all reasons are handled + for (RequestFailureReason reason : RequestFailureReason.values()) + forReason(reason); + } + + // Allow RequestFailureReason to force class load to check failure reasons are handled + public static void init() {} + + public static final IVersionedSerializer<RequestFailure> serializer = new IVersionedSerializer<RequestFailure>() + { + @Override + public void serialize(RequestFailure t, DataOutputPlus out, int version) throws IOException + { + RequestFailureReason.serializer.serialize(t.reason, out, version); + if (version >= MessagingService.VERSION_60) + nullableRemoteExceptionSerializer.serialize(t.failure, out, version); + } + + @Override + public RequestFailure deserialize(DataInputPlus in, int version) throws IOException + { + RequestFailureReason reason = RequestFailureReason.serializer.deserialize(in, version); + Throwable failure = null; + if (version >= MessagingService.VERSION_60) + failure = nullableRemoteExceptionSerializer.deserialize(in, version); + if (failure == null) + return forReason(reason); + else + return new RequestFailure(reason, failure); + } + + @Override + public long serializedSize(RequestFailure t, int version) + { + long size = RequestFailureReason.serializer.serializedSize(t.reason, version); + if (version >= MessagingService.VERSION_60) + size += nullableRemoteExceptionSerializer.serializedSize(t.failure, version); + return size; + } + }; + + @Nonnull + public final RequestFailureReason reason; + + @Nullable + public final Throwable failure; + + public static RequestFailure forException(Throwable t) + { + if (t instanceof TombstoneOverwhelmingException) + return READ_TOO_MANY_TOMBSTONES; + + if (t instanceof IncompatibleSchemaException) + return INCOMPATIBLE_SCHEMA; + + if (t instanceof NotCMSException) + return NOT_CMS; + + if (t instanceof InvalidRoutingException) + return INVALID_ROUTING; + + if (t instanceof RetryOnDifferentSystemException) + return RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM; + + if (t instanceof CoordinatorBehindException) + return COORDINATOR_BEHIND; + ++ if (t instanceof TruncateException) ++ return TRUNCATE_FAILED; ++ + return new RequestFailure(t); + } + + public static RequestFailure forReason(RequestFailureReason reason) + { + switch (reason) + { + default: throw new IllegalStateException("Unhandled request failure reason " + reason); + case UNKNOWN: return UNKNOWN; + case READ_TOO_MANY_TOMBSTONES: return READ_TOO_MANY_TOMBSTONES; + case TIMEOUT: return TIMEOUT; + case INCOMPATIBLE_SCHEMA: return INCOMPATIBLE_SCHEMA; + case READ_SIZE: return READ_SIZE; + case NODE_DOWN: return NODE_DOWN; + case NOT_CMS: return NOT_CMS; + case INVALID_ROUTING: return INVALID_ROUTING; + case INDEX_NOT_AVAILABLE: return INDEX_NOT_AVAILABLE; + case COORDINATOR_BEHIND: return COORDINATOR_BEHIND; + 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; + } + } + + private RequestFailure(RequestFailureReason reason) + { + this(reason, null); + } + + public RequestFailure(@Nonnull Throwable failure) + { + this(RequestFailureReason.UNKNOWN, failure); + } + + public RequestFailure(@Nonnull RequestFailureReason reason, @Nullable Throwable failure) + { + checkNotNull(reason); + this.reason = reason; + this.failure = failure; + } + + @Override + public String toString() + { + return "RequestFailure{" + + "reason=" + reason + + ", failure='" + failure + '\'' + + '}'; + } +} diff --cc src/java/org/apache/cassandra/exceptions/RequestFailureReason.java index 4e074c9173,99cc07be59..fbaa7b7da5 --- a/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailureReason.java @@@ -37,29 -30,15 +37,30 @@@ import static org.apache.cassandra.net. public enum RequestFailureReason { - UNKNOWN (0), - READ_TOO_MANY_TOMBSTONES (1), - TIMEOUT (2), - INCOMPATIBLE_SCHEMA (3), - READ_SIZE (4), - NODE_DOWN (5), - INDEX_NOT_AVAILABLE (6), - READ_TOO_MANY_INDEXES (7), - TRUNCATE_FAILED (12); + UNKNOWN (0), + READ_TOO_MANY_TOMBSTONES (1), + TIMEOUT (2), + INCOMPATIBLE_SCHEMA (3), + READ_SIZE (4), + // below reason is only logged, but it does not have associated exception + NODE_DOWN (5), + INDEX_NOT_AVAILABLE (6), + // below reason does not have an associated exception + READ_TOO_MANY_INDEXES (7), + NOT_CMS (8), + 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), + ; + + static + { + // Load RequestFailure class to check that all request failure reasons are handled + RequestFailure.init(); + } public static final Serializer serializer = new Serializer(); @@@ -75,39 -53,22 +76,40 @@@ static { + EnumSet<RequestFailureReason> withoutExceptions = EnumSet.of(UNKNOWN, NODE_DOWN, READ_TOO_MANY_INDEXES); + Sets.SetView<RequestFailureReason> withExceptions = Sets.difference(EnumSet.allOf(RequestFailureReason.class), withoutExceptions); RequestFailureReason[] reasons = values(); - int max = -1; - for (RequestFailureReason r : reasons) - max = max(r.code, max); - - RequestFailureReason[] codeMap = new RequestFailureReason[max + 1]; - for (RequestFailureReason reason : reasons) { - if (codeMap[reason.code] != null) + if (codeToReasonMap.put(reason.code, reason) != null) throw new RuntimeException("Two RequestFailureReason-s that map to the same code: " + reason.code); - codeMap[reason.code] = reason; } - codeToReasonMap = codeMap; + exceptionToReasonMap.put(TombstoneOverwhelmingException.class, READ_TOO_MANY_TOMBSTONES); + exceptionToReasonMap.put(WriteTimeoutException.class, TIMEOUT); + exceptionToReasonMap.put(IncompatibleSchemaException.class, INCOMPATIBLE_SCHEMA); + exceptionToReasonMap.put(ReadSizeAbortException.class, READ_SIZE); + exceptionToReasonMap.put(IndexNotAvailableException.class, INDEX_NOT_AVAILABLE); + exceptionToReasonMap.put(NotCMSException.class, NOT_CMS); + exceptionToReasonMap.put(InvalidRoutingException.class, INVALID_ROUTING); + 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()) + { + EnumSet<RequestFailureReason> actual = EnumSet.copyOf(exceptionToReasonMap.values()); + Sets.SetView<RequestFailureReason> missing = Sets.difference(withExceptions, actual); + Sets.SetView<RequestFailureReason> added = Sets.difference(actual, withExceptions); + StringBuilder sb = new StringBuilder(); + if (!missing.isEmpty()) + sb.append("Expected the following RequestFailureReason, but were missing: ").append(missing).append('\n'); + if (!added.isEmpty()) + sb.append("Unexpected RequestFailureReason found: ").append(added); + throw new AssertionError(sb.toString()); + } } public static RequestFailureReason fromCode(int code) diff --cc src/java/org/apache/cassandra/net/InboundSink.java index 734c9a1ba1,95718c5a90..6de23ede46 --- a/src/java/org/apache/cassandra/net/InboundSink.java +++ b/src/java/org/apache/cassandra/net/InboundSink.java @@@ -23,20 -22,14 +23,21 @@@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import java.util.function.Predicate; +import net.openhft.chronicle.core.util.ThrowingConsumer; + 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.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; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.tcm.NotCMSException; import org.apache.cassandra.utils.NoSpamLogger; /** @@@ -127,25 -101,14 +128,26 @@@ public class InboundSink implements Inb { fail(message.header, t); - if (t instanceof TombstoneOverwhelmingException || - t instanceof IndexNotAvailableException || - t instanceof TruncateException) + if (t instanceof NotCMSException || t instanceof CoordinatorBehindException) + { + noSpamLogger.warn(t.getMessage()); + } + else if (t instanceof TombstoneOverwhelmingException || + t instanceof IndexNotAvailableException || + t instanceof IndexBuildInProgressException || - t instanceof InvalidRoutingException) ++ t instanceof InvalidRoutingException || ++ t instanceof TruncateException) + { noSpamLogger.error(t.getMessage()); + } else if (t instanceof RuntimeException) + { throw (RuntimeException) t; + } else + { throw new RuntimeException(t); + } } } diff --cc test/unit/org/apache/cassandra/db/TruncateBlockingTest.java index 0000000000,63ac47145e..6d464a42e9 mode 000000,100644..100644 --- a/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java +++ b/test/unit/org/apache/cassandra/db/TruncateBlockingTest.java @@@ -1,0 -1,226 +1,226 @@@ + /* + * 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 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.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 org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; + import static org.junit.Assert.assertFalse; + import static org.junit.Assert.assertNotNull; + + @RunWith(BMUnitRunner.class) + public class TruncateBlockingTest extends CQLTester + { + @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()); ++ 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); ++ .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()); ++ cfs.getLiveSSTables().isEmpty()); + + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + + @Test + @BMRule(name = "no-op waitForCessation", + targetClass = "org.apache.cassandra.db.compaction.CompactionManager", + targetMethod = "waitForCessation", + action = "return;") + 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); ++ .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()); ++ 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()); ++ 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)); ++ 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()); ++ 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); ++ .as("Unable to cancel in-progress compactions. Usually retrying will work") ++ .isInstanceOf(RuntimeException.class); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + } diff --cc test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java index f707a16763,88a3987b33..3faa9b00f8 --- a/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CancelCompactionsTest.java @@@ -31,9 -31,8 +31,10 @@@ import java.util.concurrent.TimeUnit import java.util.stream.Collectors; 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; @@@ -374,6 -375,115 +377,115 @@@ public class CancelCompactionsTest exte 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()); ++ 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); + } + } + + @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()); ++ assertTrue(CompactionManager.instance.submitMaximal(cfs, false, -1).isEmpty()); + } + finally + { + CompactionManager.instance.active.finishCompaction(holder); + } + } + private List<SSTableReader> createSSTables(ColumnFamilyStore cfs, int count, int startGeneration) { List<SSTableReader> sstables = new ArrayList<>(); diff --cc test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java index 4be82d491c,0000000000..35f186ad4e mode 100644,000000..100644 --- a/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java +++ b/test/unit/org/apache/cassandra/exceptions/RequestFailureReasonTest.java @@@ -1,99 -1,0 +1,100 @@@ +/* + * 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.exceptions; + +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertEquals; + + +public class RequestFailureReasonTest +{ + private static final RequestFailureReason[] REASONS = RequestFailureReason.values(); + private static final Object[][] EXPECTED_VALUES = + { + { 0, "UNKNOWN" }, + { 1, "READ_TOO_MANY_TOMBSTONES" }, + { 2, "TIMEOUT" }, + { 3, "INCOMPATIBLE_SCHEMA" }, + { 4, "READ_SIZE" }, + { 5, "NODE_DOWN" }, + { 6, "INDEX_NOT_AVAILABLE" }, + { 7, "READ_TOO_MANY_INDEXES" }, + { 8, "NOT_CMS" }, + { 9, "INVALID_ROUTING" }, + { 10, "COORDINATOR_BEHIND" }, + { 11, "RETRY_ON_DIFFERENT_TRANSACTION_SYSTEM" }, ++ { 12, "TRUNCATE_FAILED" }, + { 503, "INDEX_BUILD_IN_PROGRESS" }, + }; + + @Test + public void testEnumCodesAndNames() + { + for (int i = 0; i < REASONS.length; i++) + { + assertEquals("RequestFailureReason code mismatch for " + + REASONS[i].name(), EXPECTED_VALUES[i][0], REASONS[i].code); + assertEquals("RequestFailureReason name mismatch for code " + + REASONS[i].code, EXPECTED_VALUES[i][1], REASONS[i].name()); + } + + assertEquals("Number of RequestFailureReason enum constants has changed. Update the test.", + EXPECTED_VALUES.length, REASONS.length); + } + + @Test + public void testFromCode() + { + // Test valid codes + for (Object[] expected : EXPECTED_VALUES) + { + int code = (Integer) expected[0]; + String name = (String) expected[1]; + assertEquals(RequestFailureReason.valueOf(name), RequestFailureReason.fromCode(code)); + } + + // Test invalid codes + assertEquals(RequestFailureReason.UNKNOWN, RequestFailureReason.fromCode(200)); + assertEquals(RequestFailureReason.UNKNOWN, RequestFailureReason.fromCode(999)); + assertThatThrownBy(() -> RequestFailureReason.fromCode(-1)).isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testExceptionSubclassMapping() + { + // Create a subclass of UnknownTableException + class CustomUnknownTableException extends IncompatibleSchemaException + { + public CustomUnknownTableException(String ks) + { + super(ks); + } + } + + // Verify the parent class still maps correctly + assertEquals(RequestFailureReason.INCOMPATIBLE_SCHEMA, + RequestFailureReason.forException(new CustomUnknownTableException("ks"))); + + // Test unmapped exception returns UNKNOWN + assertEquals(RequestFailureReason.UNKNOWN, + RequestFailureReason.forException(new RuntimeException("test"))); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
