From 7f0e5f8e8bfc747a89ff6bdcf18d40ad96af4c7b Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Mon, 27 Jan 2025 16:27:06 +0000 Subject: [PATCH] Follow-up to CASSANDRA-20228: - Fix waiting state callback computes different route to initiator - Invariants.checkX -> Invariants.requireX (to allow complementary Invariants.expectX as appropriate) patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20228 Follow-up to CASSANDRA-20228: - Fix AccordUpdateParameters to correctly supply List cell paths derived from applyAt - Fix ExecuteAtSerialize.serialiseNullable - Fix topologies flush regression caused by markRetired forcing a flush on every call patch by Benedict; reviewed by Ariel Weisberg for CASSANDRA-20228 --- modules/accord | 2 +- .../apache/cassandra/config/AccordSpec.java | 2 +- .../cassandra/cql3/UpdateParameters.java | 17 +----- .../statements/ModificationStatement.java | 23 ++++--- .../apache/cassandra/cql3/terms/Lists.java | 45 +------------- .../db/compaction/CompactionIterator.java | 6 +- .../db/marshal/ByteArrayAccessor.java | 2 +- .../cassandra/db/marshal/CompositeType.java | 4 +- .../db/virtual/AccordDebugKeyspace.java | 2 +- .../cassandra/dht/AccordBytesSplitter.java | 12 ++-- .../dht/LocalCompositePrefixPartitioner.java | 6 +- .../dht/OrderPreservingPartitioner.java | 6 +- .../cassandra/journal/EntrySerializer.java | 8 +-- .../org/apache/cassandra/journal/Flusher.java | 4 +- .../org/apache/cassandra/journal/Journal.java | 12 ++-- .../apache/cassandra/journal/OnDiskIndex.java | 2 +- .../org/apache/cassandra/journal/Segment.java | 10 ++-- .../apache/cassandra/journal/Segments.java | 12 ++-- .../org/apache/cassandra/schema/TableId.java | 4 +- .../cassandra/schema/TableMetadata.java | 4 +- .../cassandra/service/accord/AccordCache.java | 36 +++++------ .../service/accord/AccordCacheEntry.java | 36 +++++------ .../service/accord/AccordCommandStore.java | 12 ++-- .../accord/AccordConfigurationService.java | 4 +- .../service/accord/AccordDataStore.java | 6 +- .../service/accord/AccordExecutor.java | 32 +++++----- .../accord/AccordExecutorInfiniteLoops.java | 2 +- .../service/accord/AccordExecutorSimple.java | 2 +- .../accord/AccordFastPathCoordinator.java | 2 +- .../accord/AccordFetchCoordinator.java | 22 +++---- .../service/accord/AccordJournal.java | 24 ++++---- .../service/accord/AccordJournalTable.java | 4 +- .../accord/AccordJournalValueSerializers.java | 4 +- .../service/accord/AccordKeyspace.java | 30 ++++------ .../accord/AccordSegmentCompactor.java | 8 +-- .../service/accord/AccordService.java | 10 ++-- .../service/accord/AccordSyncPropagator.java | 2 +- .../cassandra/service/accord/AccordTask.java | 30 +++++----- .../service/accord/AccordTopology.java | 6 +- .../service/accord/CommandsForRanges.java | 2 +- .../service/accord/EndpointMapping.java | 8 +-- .../cassandra/service/accord/JournalKey.java | 4 +- .../cassandra/service/accord/TokenRange.java | 2 +- .../service/accord/api/AccordAgent.java | 2 +- .../ParameterizedFastPathStrategy.java | 4 +- .../fastpath/SimpleFastPathStrategy.java | 2 +- .../accord/interop/AccordInteropApply.java | 4 +- .../interop/AccordInteropExecution.java | 8 +-- .../accord/interop/AccordInteropPersist.java | 4 +- .../interop/AccordInteropReadCallback.java | 2 +- .../accord/serializers/AcceptSerializers.java | 2 +- .../accord/serializers/ApplySerializers.java | 2 +- .../accord/serializers/AwaitSerializer.java | 4 +- .../serializers/CommandSerializers.java | 12 ++-- .../serializers/CommandStoreSerializers.java | 2 +- .../serializers/SmallEnumSerializer.java | 2 +- .../serializers/WaitingOnSerializer.java | 10 ++-- .../accord/txn/AccordUpdateParameters.java | 34 ++++++++--- .../cassandra/service/accord/txn/TxnData.java | 6 +- .../service/accord/txn/TxnUpdate.java | 4 +- .../tcm/AtomicLongBackedProcessor.java | 2 +- .../org/apache/cassandra/tcm/Processor.java | 2 +- .../apache/cassandra/tcm/log/LogReader.java | 2 +- .../apache/cassandra/tcm/log/LogState.java | 2 +- .../cassandra/tcm/membership/Directory.java | 2 +- .../apache/cassandra/utils/MergeIterator.java | 2 +- .../apache/cassandra/utils/btree/BTree.java | 4 +- .../utils/concurrent/LockWithAsyncSignal.java | 4 +- .../distributed/impl/Coordinator.java | 2 +- .../distributed/test/IntegrationTestBase.java | 2 +- .../fuzz/topology/TopologyMixupTestBase.java | 60 +++++++++---------- .../dsl/SingleOperationVisitBuilder.java | 6 +- .../execution/CQLTesterVisitExecutor.java | 4 +- .../harry/execution/CQLVisitExecutor.java | 4 +- .../harry/execution/DataTracker.java | 4 +- .../execution/InJvmDTestVisitExecutor.java | 4 +- .../execution/QueryBuildingVisitExecutor.java | 4 +- .../RingAwareInJvmDTestVisitExecutor.java | 2 +- .../cassandra/harry/gen/Generators.java | 2 +- .../harry/gen/InvertibleGenerator.java | 18 +++--- .../harry/model/QuiescentChecker.java | 2 +- .../apache/cassandra/harry/op/Operations.java | 2 +- .../harry/test/SimpleBijectionTest.java | 2 +- .../simulator/logging/RunStartDefiner.java | 2 +- .../simulator/test/HarrySimulatorTest.java | 2 +- .../simulator/test/HarryValidatingQuery.java | 4 +- .../cassandra/concurrent/ManualExecutor.java | 2 +- .../apache/cassandra/cql3/ast/Visitor.java | 2 +- .../apache/cassandra/journal/IndexTest.java | 2 +- .../service/accord/EpochSyncTest.java | 18 +++--- .../service/accord/MockDiskStateManager.java | 2 +- .../SimulatedAccordCommandStoreTestBase.java | 2 +- .../cassandra/utils/SimulatedMiniCluster.java | 2 +- 93 files changed, 353 insertions(+), 398 deletions(-) diff --git a/modules/accord b/modules/accord index 41aacd31c9..c7a789b1f4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 41aacd31c93251043154b5f73239345ee68b7957 +Subproject commit c7a789b1f424771a4befab6bcb91edd4ab5d7198 diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index f062c9a570..720a2db772 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -272,7 +272,7 @@ public class AccordSpec return units.convert(nanos, TimeUnit.NANOSECONDS); long flushPeriodNanos = flushPeriod(TimeUnit.NANOSECONDS); - Invariants.checkState(flushPeriodNanos > 0); + Invariants.require(flushPeriodNanos > 0); nanos = periodicFlushLagBlock.to(TimeUnit.NANOSECONDS) + flushPeriodNanos; // it is possible for this to race and cache the wrong value after an update flushCombinedBlockPeriod = nanos; diff --git a/src/java/org/apache/cassandra/cql3/UpdateParameters.java b/src/java/org/apache/cassandra/cql3/UpdateParameters.java index 8b0e5cdf7c..6ef6c7bb6c 100644 --- a/src/java/org/apache/cassandra/cql3/UpdateParameters.java +++ b/src/java/org/apache/cassandra/cql3/UpdateParameters.java @@ -51,10 +51,9 @@ public class UpdateParameters public final TableMetadata metadata; public final ClientState clientState; public final QueryOptions options; - public final boolean constructingAccordBaseUpdate; private final long nowInSec; - private final long timestamp; + protected final long timestamp; private final int ttl; private final DeletionTime deletionTime; @@ -75,18 +74,6 @@ public class UpdateParameters long nowInSec, int ttl, Map prefetchedRows) throws InvalidRequestException - { - this(metadata, clientState, options, timestamp, nowInSec, ttl, prefetchedRows, false); - } - - public UpdateParameters(TableMetadata metadata, - ClientState clientState, - QueryOptions options, - long timestamp, - long nowInSec, - int ttl, - Map prefetchedRows, - boolean constructingAccordBaseUpdate) throws InvalidRequestException { this.metadata = metadata; this.clientState = clientState; @@ -104,8 +91,6 @@ public class UpdateParameters // it to avoid potential confusion. if (timestamp == Long.MIN_VALUE) throw new InvalidRequestException(String.format("Out of bound timestamp, must be in [%d, %d]", Long.MIN_VALUE + 1, Long.MAX_VALUE)); - - this.constructingAccordBaseUpdate = constructingAccordBaseUpdate; } public void newRow(Clustering clustering) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 8dc2624d2c..a02cd27b4e 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -969,8 +969,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa local, timestamp, nowInSeconds, - requestTime, - constructingAccordBaseUpdate); + requestTime + ); for (ByteBuffer key : keys) { Validation.validateKey(metadata(), key); @@ -994,7 +994,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa if (restrictions.hasClusteringColumnsRestrictions() && clusterings.isEmpty()) return; - UpdateParameters params = makeUpdateParameters(keys, clusterings, state, options, local, timestamp, nowInSeconds, requestTime, constructingAccordBaseUpdate); + UpdateParameters params = makeUpdateParameters(keys, clusterings, state, options, local, timestamp, nowInSeconds, requestTime); for (ByteBuffer key : keys) { @@ -1052,8 +1052,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa boolean local, long timestamp, long nowInSeconds, - Dispatcher.RequestTime requestTime, - boolean constructingAccordBaseUpdate) + Dispatcher.RequestTime requestTime) { if (clusterings.contains(Clustering.STATIC_CLUSTERING)) return makeUpdateParameters(keys, @@ -1064,8 +1063,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa local, timestamp, nowInSeconds, - requestTime, - constructingAccordBaseUpdate); + requestTime + ); return makeUpdateParameters(keys, new ClusteringIndexNamesFilter(clusterings, false), @@ -1075,8 +1074,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa local, timestamp, nowInSeconds, - requestTime, - constructingAccordBaseUpdate); + requestTime + ); } private UpdateParameters makeUpdateParameters(Collection keys, @@ -1087,8 +1086,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa boolean local, long timestamp, long nowInSeconds, - Dispatcher.RequestTime requestTime, - boolean constructingAccordBaseUpdate) + Dispatcher.RequestTime requestTime) { // Some lists operation requires reading Map lists = @@ -1106,8 +1104,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa getTimestamp(timestamp, options), nowInSeconds, getTimeToLive(options), - lists, - constructingAccordBaseUpdate); + lists); } public static abstract class Parsed extends QualifiedStatement diff --git a/src/java/org/apache/cassandra/cql3/terms/Lists.java b/src/java/org/apache/cassandra/cql3/terms/Lists.java index 01b29c4a69..579b783d8d 100644 --- a/src/java/org/apache/cassandra/cql3/terms/Lists.java +++ b/src/java/org/apache/cassandra/cql3/terms/Lists.java @@ -44,7 +44,6 @@ import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.MultiElementType; -import org.apache.cassandra.db.marshal.TimeUUIDType; import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.db.rows.CellPath; import org.apache.cassandra.db.rows.ComplexColumnData; @@ -52,7 +51,6 @@ import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.TimeUUID; import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse; import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest; @@ -68,13 +66,6 @@ public abstract class Lists @SuppressWarnings("unused") private static final Logger logger = LoggerFactory.getLogger(Lists.class); - /** - * Sentinel value indicating the cell path should be replaced by Accord with one based on the transaction executeAt - */ - private static final TimeUUID ACCORD_CELL_PATH_SENTINEL_UUID = TimeUUID.atUnixMicrosWithLsb(0, 0); - public static final CellPath ACCORD_DUMMY_CELL_PATH = CellPath.create(ACCORD_CELL_PATH_SENTINEL_UUID.toBytes()); - private static final long ACCORD_CELL_PATH_SENTINEL_MSB = ACCORD_CELL_PATH_SENTINEL_UUID.msb(); - private Lists() {} public static ColumnSpecification indexSpecOf(ColumnSpecification column) @@ -160,33 +151,6 @@ public abstract class Lists return type == null ? null : ListType.getInstance(type, false); } - /** - * Return a function that given a cell with an ACCORD_CELL_PATH_SENTINEL_MSB will - * return a new CellPath with a TimeUUID that increases monotonically every time it is called or - * the existing cell path if path does not contain ACCORD_CELL_PATH_SENTINEL_MSB. - * - * Only intended to work with list cell paths where list append needs a timestamp based on the executeAt - * of the Accord transaction appending the cell. - * @param timestampMicros executeAt timestamp to use as the MSB for generated cell paths - */ - public static com.google.common.base.Function accordListPathSupplier(long timestampMicros) - { - return new com.google.common.base.Function() - { - final long timeUuidMsb = TimeUUID.unixMicrosToMsb(timestampMicros); - long cellIndex = 0; - @Override - public CellPath apply(Cell cell) - { - CellPath path = cell.path(); - if (ACCORD_CELL_PATH_SENTINEL_MSB == path.get(0).getLong(0)) - return CellPath.create(ByteBuffer.wrap(TimeUUID.toBytes(timeUuidMsb, TimeUUIDType.signedBytesToNativeLong(cellIndex++)))); - else - return path; - } - }; - } - public static class Literal extends Term.Raw { private final List elements; @@ -463,17 +427,10 @@ public abstract class Lists // during SSTable write. Guardrails.itemsPerCollection.guard(type.collectionSize(elements), column.name.toString(), false, params.clientState); - long cellIndex = 0; int dataSize = 0; for (ByteBuffer buffer : elements) { - ByteBuffer cellPath; - // Accord will need to replace this value later once it knows the executeAt timestamp - // so just put a TimeUUID with MSB sentinel for now - if (params.constructingAccordBaseUpdate) - cellPath = TimeUUID.atUnixMicrosWithLsb(0, cellIndex++).toBytes(); - else - cellPath = ByteBuffer.wrap(params.nextTimeUUIDAsBytes()); + ByteBuffer cellPath = ByteBuffer.wrap(params.nextTimeUUIDAsBytes()); Cell cell = params.addCell(column, CellPath.create(cellPath), buffer); dataSize += cell.dataSize(); } diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 1f3af90431..43b5327c7d 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -905,7 +905,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte AccordJournal.Builder commandBuilder = (AccordJournal.Builder) builder; if (commandBuilder.isEmpty()) { - Invariants.checkState(rows.isEmpty()); + Invariants.require(rows.isEmpty()); return partition; } @@ -954,9 +954,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte if (lastOffset != -1) { - Invariants.checkState(descriptor <= lastDescriptor, + Invariants.require(descriptor <= lastDescriptor, "Descriptors were accessed out of order: %d was accessed after %d", descriptor, lastDescriptor); - Invariants.checkState(descriptor != lastDescriptor || + Invariants.require(descriptor != lastDescriptor || offset < lastOffset, "Offsets within %s were accessed out of order: %d was accessed after %s", offset, lastOffset); } diff --git a/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java b/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java index 01b1de63e1..7035a25147 100644 --- a/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java +++ b/src/java/org/apache/cassandra/db/marshal/ByteArrayAccessor.java @@ -108,7 +108,7 @@ public class ByteArrayAccessor implements ValueAccessor @Override public byte[] slice(byte[] input, int offset, int length) { - Invariants.checkArgument(offset + length <= input.length); + Invariants.requireArgument(offset + length <= input.length); return Arrays.copyOfRange(input, offset, offset + length); } diff --git a/src/java/org/apache/cassandra/db/marshal/CompositeType.java b/src/java/org/apache/cassandra/db/marshal/CompositeType.java index e175045049..78ef0694ef 100644 --- a/src/java/org/apache/cassandra/db/marshal/CompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/CompositeType.java @@ -312,9 +312,9 @@ public class CompositeType extends AbstractCompositeType byte[] result = Arrays.copyOf(bytes, c); if (bytes != tmpBytes) tmpFlattenBuffer.set(bytes); byte[] test = super.asFlatComparableBytes(accessor, data, version); - if (Invariants.isParanoid() && Invariants.testParanoia(LINEAR, CONSTANT, LOW)) Invariants.checkState(Arrays.equals(test, result)); + if (Invariants.isParanoid() && Invariants.testParanoia(LINEAR, CONSTANT, LOW)) Invariants.require(Arrays.equals(test, result)); V roundtrip = fromComparableBytes(accessor, ByteSource.peekable(ByteSource.of(result, version)), version); - Invariants.checkState(accessor.compare(data, roundtrip, accessor) == 0); + Invariants.require(accessor.compare(data, roundtrip, accessor) == 0); return result; } diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index d8b81b866f..757fb9488f 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -578,7 +578,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace CommandStoreTxnBlockedGraph.TxnState txn = shard.txns.get(txnId); if (txn == null) { - Invariants.checkState(reason == Reason.Self, "Txn %s unknown for reason %s", txnId, reason); + Invariants.require(reason == Reason.Self, "Txn %s unknown for reason %s", txnId, reason); return; } // was it applied? If so ignore it diff --git a/src/java/org/apache/cassandra/dht/AccordBytesSplitter.java b/src/java/org/apache/cassandra/dht/AccordBytesSplitter.java index 66eb6b73fc..b35724f703 100644 --- a/src/java/org/apache/cassandra/dht/AccordBytesSplitter.java +++ b/src/java/org/apache/cassandra/dht/AccordBytesSplitter.java @@ -26,7 +26,7 @@ import accord.utils.Invariants; import org.apache.cassandra.service.accord.api.AccordRoutingKey; import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey; -import static accord.utils.Invariants.checkArgument; +import static accord.utils.Invariants.requireArgument; import static java.math.BigInteger.ONE; import static java.math.BigInteger.ZERO; @@ -50,9 +50,9 @@ public class AccordBytesSplitter extends AccordSplitter // Since BOP is already not working/supported I think it's fine to punt on this. if (bytesLength == 0) { - checkArgument(ranges.size() <= 1); - checkArgument(ranges.isEmpty() || ranges.get(0).start().getClass() == SentinelKey.class); - checkArgument(ranges.isEmpty() || ranges.get(0).end().getClass() == SentinelKey.class); + requireArgument(ranges.size() <= 1); + requireArgument(ranges.isEmpty() || ranges.get(0).start().getClass() == SentinelKey.class); + requireArgument(ranges.isEmpty() || ranges.get(0).end().getClass() == SentinelKey.class); // Intentionally does not match 16 that is used by ServerTestUtils.getRandomToken to elicit breakage bytesLength = 8; } @@ -75,7 +75,7 @@ public class AccordBytesSplitter extends AccordSplitter BigInteger valueForToken(Token token) { byte[] bytes = ((ByteOrderedPartitioner.BytesToken) token).token; - checkArgument(bytes.length <= byteLength); + requireArgument(bytes.length <= byteLength); BigInteger value = ZERO; for (int i = 0 ; i < bytes.length ; ++i) value = value.add(BigInteger.valueOf(bytes[i] & 0xffL).shiftLeft((byteLength - 1 - i) * 8)); @@ -85,7 +85,7 @@ public class AccordBytesSplitter extends AccordSplitter @Override Token tokenForValue(BigInteger value) { - Invariants.checkArgument(value.compareTo(ZERO) >= 0); + Invariants.requireArgument(value.compareTo(ZERO) >= 0); byte[] bytes = new byte[byteLength]; for (int i = 0 ; i < bytes.length ; ++i) bytes[i] = value.shiftRight((byteLength - 1 - i) * 8).byteValue(); diff --git a/src/java/org/apache/cassandra/dht/LocalCompositePrefixPartitioner.java b/src/java/org/apache/cassandra/dht/LocalCompositePrefixPartitioner.java index c0a056e043..10067a7cac 100644 --- a/src/java/org/apache/cassandra/dht/LocalCompositePrefixPartitioner.java +++ b/src/java/org/apache/cassandra/dht/LocalCompositePrefixPartitioner.java @@ -81,7 +81,7 @@ public class LocalCompositePrefixPartitioner extends LocalPartitioner @Override public int compareTo(Token o) { - Invariants.checkArgument(o instanceof AbstractCompositePrefixToken); + Invariants.requireArgument(o instanceof AbstractCompositePrefixToken); AbstractCompositePrefixToken that = (AbstractCompositePrefixToken) o; CompositeType comparator = comparatorForPrefixLength(Math.min(this.prefixSize(), that.prefixSize())); return comparator.compare(this.token, that.token); @@ -136,7 +136,7 @@ public class LocalCompositePrefixPartitioner extends LocalPartitioner public PrefixToken(ByteBuffer token, int prefixSize) { super(token); - Invariants.checkArgument(prefixSize > 0); + Invariants.requireArgument(prefixSize > 0); this.prefixSize = prefixSize; } @@ -231,7 +231,7 @@ public class LocalCompositePrefixPartitioner extends LocalPartitioner public DecoratedKey decoratedKey(Object... values) { - Invariants.checkArgument(values.length == prefixComparators.size()); + Invariants.requireArgument(values.length == prefixComparators.size()); ByteBuffer key = createPrefixKey(values); return decorateKey(key); } diff --git a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java index ea38af129f..87968321b2 100644 --- a/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java +++ b/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java @@ -48,7 +48,7 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.ObjectSizes; import org.apache.cassandra.utils.Pair; -import static accord.utils.Invariants.checkArgument; +import static accord.utils.Invariants.requireArgument; import static java.lang.Integer.max; import static java.math.BigInteger.ONE; import static java.math.BigInteger.ZERO; @@ -330,7 +330,7 @@ public class OrderPreservingPartitioner implements IPartitioner BigInteger valueForToken(Token token) { String chars = ((StringToken) token).token; - checkArgument(chars.length() <= charLength); + requireArgument(chars.length() <= charLength); BigInteger value = ZERO; for (int i = 0 ; i < chars.length() ; ++i) value = value.add(BigInteger.valueOf(chars.charAt(i) & 0xffffL).shiftLeft((charLength - 1 - i) * 16)); @@ -341,7 +341,7 @@ public class OrderPreservingPartitioner implements IPartitioner Token tokenForValue(BigInteger value) { // TODO (required): test - checkArgument(value.compareTo(ZERO) >= 0); + requireArgument(value.compareTo(ZERO) >= 0); char[] chars = new char[charLength]; for (int i = 0 ; i < chars.length ; ++i) chars[i] = (char) value.shiftRight((charLength - 1 - i) * 16).shortValue(); diff --git a/src/java/org/apache/cassandra/journal/EntrySerializer.java b/src/java/org/apache/cassandra/journal/EntrySerializer.java index d991e5088e..bf6464bb68 100644 --- a/src/java/org/apache/cassandra/journal/EntrySerializer.java +++ b/src/java/org/apache/cassandra/journal/EntrySerializer.java @@ -40,8 +40,8 @@ public final class EntrySerializer { int start = out.position(); int totalSize = out.getInt() - start; - Invariants.checkState(totalSize == TypeSizes.INT_SIZE + out.remaining()); - Invariants.checkState(totalSize == headerSize(keySupport, userVersion) + record.remaining() + TypeSizes.INT_SIZE); + Invariants.require(totalSize == TypeSizes.INT_SIZE + out.remaining()); + Invariants.require(totalSize == headerSize(keySupport, userVersion) + record.remaining() + TypeSizes.INT_SIZE); keySupport.serialize(key, out, userVersion); @@ -50,7 +50,7 @@ public final class EntrySerializer int recordSize = record.remaining(); int recordEnd = out.position() + recordSize; - Invariants.checkState(out.limit() == recordEnd + TypeSizes.INT_SIZE); + Invariants.require(out.limit() == recordEnd + TypeSizes.INT_SIZE); ByteBufferUtil.copyBytes(record, record.position(), out, out.position(), recordSize); // update and write crcs @@ -77,7 +77,7 @@ public final class EntrySerializer int start = from.position(); { int totalSize = from.getInt(start) - start; - Invariants.checkState(totalSize == from.remaining()); + Invariants.require(totalSize == from.remaining()); CRC32 crc = Crc.crc32(); int headerSize = EntrySerializer.headerSize(keySupport, userVersion); diff --git a/src/java/org/apache/cassandra/journal/Flusher.java b/src/java/org/apache/cassandra/journal/Flusher.java index 2aba68d8ea..fb67cf26cb 100644 --- a/src/java/org/apache/cassandra/journal/Flusher.java +++ b/src/java/org/apache/cassandra/journal/Flusher.java @@ -275,7 +275,7 @@ final class Flusher if (flushPeriodNanos <= 0) { - Invariants.checkState(params.flushMode() != PERIODIC); + Invariants.require(params.flushMode() != PERIODIC); haveWork.acquire(1); } else @@ -517,7 +517,7 @@ final class Flusher signal.awaitThrowUncheckedOnInterrupt(); Journal.State state = journal.state.get(); - Invariants.checkState(state == Journal.State.NORMAL, + Invariants.require(state == Journal.State.NORMAL, "Thread %s outlived journal, which is in %s state", Thread.currentThread(), state); } else diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 8fd4d80ff9..cd7bb53d07 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -197,7 +197,7 @@ public class Journal implements Shutdownable public void start() { - Invariants.checkState(state.compareAndSet(State.UNINITIALIZED, State.INITIALIZING), + Invariants.require(state.compareAndSet(State.UNINITIALIZED, State.INITIALIZING), "Unexpected journal state during initialization", state); metrics.register(flusher); @@ -216,7 +216,7 @@ public class Journal implements Shutdownable releaser = executorFactory().sequential(name + "-releaser"); allocator = executorFactory().infiniteLoop(name + "-allocator", new AllocateRunnable(), SAFE, NON_DAEMON, SYNCHRONIZED); advanceSegment(null); - Invariants.checkState(state.compareAndSet(State.INITIALIZING, State.NORMAL), + Invariants.require(state.compareAndSet(State.INITIALIZING, State.NORMAL), "Unexpected journal state after initialization", state); flusher.start(); compactor.start(); @@ -252,7 +252,7 @@ public class Journal implements Shutdownable { try { - Invariants.checkState(state.compareAndSet(State.NORMAL, State.SHUTDOWN), + Invariants.require(state.compareAndSet(State.NORMAL, State.SHUTDOWN), "Unexpected journal state while trying to shut down", state); allocator.shutdown(); wakeAllocator(); // Wake allocator to force it into shutdown @@ -268,7 +268,7 @@ public class Journal implements Shutdownable releaser.awaitTermination(1, TimeUnit.MINUTES); closeAllSegments(); metrics.deregister(); - Invariants.checkState(state.compareAndSet(State.SHUTDOWN, State.TERMINATED), + Invariants.require(state.compareAndSet(State.SHUTDOWN, State.TERMINATED), "Unexpected journal state while trying to shut down", state); } catch (InterruptedException e) @@ -774,7 +774,7 @@ public class Journal implements Shutdownable else { Segment segment = segments().get(timestamp); - Invariants.checkState(segment != null, "Segment %d expected to be found, but neither current segment %d nor in active segments", timestamp, currentSegmentTimestamp); + Invariants.require(segment != null, "Segment %d expected to be found, but neither current segment %d nor in active segments", timestamp, currentSegmentTimestamp); if (segment == null) throw new IllegalArgumentException("Request the active segment " + timestamp + " but this segment does not exist"); if (!segment.isActive()) @@ -965,7 +965,7 @@ public class Journal implements Shutdownable StaticSegment.KeyOrderReader next = readers.peek(); if (next == null || !next.key().equals(key)) break; - Invariants.checkState(next == readers.poll()); + Invariants.require(next == readers.poll()); reader.accept(next.descriptor.timestamp, next.offset, next.key(), next.record(), next.descriptor.userVersion); if (next.advance()) diff --git a/src/java/org/apache/cassandra/journal/OnDiskIndex.java b/src/java/org/apache/cassandra/journal/OnDiskIndex.java index 4994e3a087..17b8a7b791 100644 --- a/src/java/org/apache/cassandra/journal/OnDiskIndex.java +++ b/src/java/org/apache/cassandra/journal/OnDiskIndex.java @@ -171,7 +171,7 @@ final class OnDiskIndex extends Index if (prev != -1) { long tmp = prev; - Invariants.checkState(readOffset(offsetAndSize) < readOffset(prev), + Invariants.require(readOffset(offsetAndSize) < readOffset(prev), () -> String.format("Offsets should be strictly reverse monotonic, but found %d following %d", readOffset(offsetAndSize), readOffset(tmp))); } diff --git a/src/java/org/apache/cassandra/journal/Segment.java b/src/java/org/apache/cassandra/journal/Segment.java index 5faee3dc41..3854f0ee27 100644 --- a/src/java/org/apache/cassandra/journal/Segment.java +++ b/src/java/org/apache/cassandra/journal/Segment.java @@ -91,7 +91,7 @@ public abstract class Segment implements SelfRefCounted>, Co int size = Index.readSize(offsetAndSize); if (read(offset, size, into)) { - Invariants.checkState(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); + Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); consumer.accept(descriptor.timestamp, offset, id, into.value, descriptor.userVersion); return true; } @@ -103,7 +103,7 @@ public abstract class Segment implements SelfRefCounted>, Co long offsetAndSize = index().lookUpLast(id); if (offsetAndSize == -1 || !read(Index.readOffset(offsetAndSize), Index.readSize(offsetAndSize), into)) return false; - Invariants.checkState(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); + Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); return true; } @@ -115,9 +115,9 @@ public abstract class Segment implements SelfRefCounted>, Co { int offset = Index.readOffset(all[i]); int size = Index.readSize(all[i]); - Invariants.checkState(offset < prevOffset); - Invariants.checkState(read(offset, size, into), "Read should always return true"); - Invariants.checkState(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); + Invariants.require(offset < prevOffset); + Invariants.require(read(offset, size, into), "Read should always return true"); + Invariants.require(id.equals(into.key), "Index for %s read incorrect key: expected %s but read %s", descriptor, id, into.key); onEntry.accept(descriptor.timestamp, offset, into.key, into.value, into.userVersion); } } diff --git a/src/java/org/apache/cassandra/journal/Segments.java b/src/java/org/apache/cassandra/journal/Segments.java index 7d63b86481..cb38a489ca 100644 --- a/src/java/org/apache/cassandra/journal/Segments.java +++ b/src/java/org/apache/cassandra/journal/Segments.java @@ -58,7 +58,7 @@ class Segments { Long2ObjectHashMap> newSegments = new Long2ObjectHashMap<>(segments); Segment oldValue = newSegments.put(activeSegment.descriptor.timestamp, activeSegment); - Invariants.checkState(oldValue == null); + Invariants.require(oldValue == null); return new Segments<>(newSegments); } @@ -66,16 +66,16 @@ class Segments { Long2ObjectHashMap> newSegments = new Long2ObjectHashMap<>(segments); Segment oldValue = segments.remove(activeSegment.descriptor.timestamp); - Invariants.checkState(oldValue.asActive().isEmpty()); + Invariants.require(oldValue.asActive().isEmpty()); return new Segments<>(newSegments); } Segments withCompletedSegment(ActiveSegment activeSegment, StaticSegment staticSegment) { - Invariants.checkArgument(activeSegment.descriptor.equals(staticSegment.descriptor)); + Invariants.requireArgument(activeSegment.descriptor.equals(staticSegment.descriptor)); Long2ObjectHashMap> newSegments = new Long2ObjectHashMap<>(segments); Segment oldValue = newSegments.put(staticSegment.descriptor.timestamp, staticSegment); - Invariants.checkState(oldValue == activeSegment, () -> String.format("old value %s != new %s", oldValue, activeSegment)); + Invariants.require(oldValue == activeSegment, () -> String.format("old value %s != new %s", oldValue, activeSegment)); return new Segments<>(newSegments); } @@ -85,13 +85,13 @@ class Segments for (StaticSegment oldSegment : oldSegments) { Segment oldValue = newSegments.remove(oldSegment.descriptor.timestamp); - Invariants.checkState(oldValue == oldSegment); + Invariants.require(oldValue == oldSegment); } for (StaticSegment compactedSegment : compactedSegments) { Segment oldValue = newSegments.put(compactedSegment.descriptor.timestamp, compactedSegment); - Invariants.checkState(oldValue == null); + Invariants.require(oldValue == null); } return new Segments<>(newSegments); diff --git a/src/java/org/apache/cassandra/schema/TableId.java b/src/java/org/apache/cassandra/schema/TableId.java index 1f4b5a5837..c1badab05d 100644 --- a/src/java/org/apache/cassandra/schema/TableId.java +++ b/src/java/org/apache/cassandra/schema/TableId.java @@ -272,7 +272,7 @@ public final class TableId implements Comparable long compact = in.readUnsignedVInt(); if (compact > 0) return fromLong(compact - 1); - Invariants.checkState(compact == 0); + Invariants.require(compact == 0); return deserialize(in); } @@ -281,7 +281,7 @@ public final class TableId implements Comparable long compact = accessor.getUnsignedVInt(src, offset); if (compact > 0) return fromLong(compact - 1); - Invariants.checkState(compact == 0); + Invariants.require(compact == 0); return deserialize(src, accessor, offset + 1); } diff --git a/src/java/org/apache/cassandra/schema/TableMetadata.java b/src/java/org/apache/cassandra/schema/TableMetadata.java index c6a6f687f2..70ebc45abf 100644 --- a/src/java/org/apache/cassandra/schema/TableMetadata.java +++ b/src/java/org/apache/cassandra/schema/TableMetadata.java @@ -82,7 +82,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.github.jamm.Unmetered; -import static accord.utils.Invariants.checkState; +import static accord.utils.Invariants.require; import static com.google.common.collect.Iterables.any; import static com.google.common.collect.Iterables.transform; import static java.lang.String.format; @@ -619,7 +619,7 @@ public class TableMetadata implements SchemaElement } } - checkState((params.transactionalMode == TransactionalMode.off && params.transactionalMigrationFrom == TransactionalMigrationFromMode.none) || !isCounter(), "Counters are not supported with Accord for table " + this); + require((params.transactionalMode == TransactionalMode.off && params.transactionalMigrationFrom == TransactionalMigrationFromMode.none) || !isCounter(), "Counters are not supported with Accord for table " + this); } /** diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java index 78fb502c59..7b7a4390fd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java @@ -66,8 +66,8 @@ import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement; import org.apache.cassandra.utils.ObjectSizes; -import static accord.utils.Invariants.checkState; import static accord.utils.Invariants.illegalState; +import static accord.utils.Invariants.require; import static org.apache.cassandra.net.MessagingService.current_version; import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED; import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.LOADED; @@ -225,7 +225,7 @@ public class AccordCache implements CacheSize @VisibleForTesting private void shrinkOrEvict(AccordCacheEntry node) { - checkState(node.references() == 0); + require(node.references() == 0); if (shrinkingOn && node.tryShrink()) { @@ -243,7 +243,7 @@ public class AccordCache implements CacheSize @VisibleForTesting public void tryEvict(AccordCacheEntry node) { - checkState(node.references() == 0); + require(node.references() == 0); if (node.isNoEvict()) { @@ -278,7 +278,7 @@ public class AccordCache implements CacheSize if (logger.isTraceEnabled()) logger.trace("Evicting {}", node); - checkState(node.isUnqueued()); + require(node.isUnqueued()); if (updateUnreferenced) { @@ -296,8 +296,8 @@ public class AccordCache implements CacheSize owner.validateLoadEvicted(node); AccordCacheEntry self = node.owner.cache.remove(node.key()); - Invariants.checkState(self.references() == 0); - checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self); + Invariants.require(self.references() == 0); + require(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self); node.notifyListeners(Listener::onEvict); node.evicted(); } @@ -316,10 +316,10 @@ public class AccordCache implements CacheSize void failedToLoad(AccordCacheEntry node) { - Invariants.checkState(node.references() == 0); + Invariants.require(node.references() == 0); if (node.isUnqueued()) { - Invariants.checkState(node.status() == EVICTED); + Invariants.require(node.status() == EVICTED); return; } node.unlink(); @@ -422,14 +422,14 @@ public class AccordCache implements CacheSize public S acquire(AccordCacheEntry node) { - Invariants.checkState(node.owner == this); + Invariants.require(node.owner == this); acquireExisting(node, false); return adapter.safeRef(node); } public void recordPreAcquired(AccordSafeState ref) { - Invariants.checkState(ref.global().owner == this); + Invariants.require(ref.global().owner == this); incrementCacheHits(); } @@ -456,7 +456,7 @@ public class AccordCache implements CacheSize Object prev = cache.put(key, node); node.initSize(parent()); - Invariants.checkState(prev == null, "%s not absent from cache: %s already present", key, node); + Invariants.require(prev == null, "%s not absent from cache: %s already present", key, node); ++size; node.notifyListeners(Listener::onAdd); maybeShrinkOrEvictSomeNodes(); @@ -494,11 +494,11 @@ public class AccordCache implements CacheSize AccordCacheEntry node = cache.get(key); - checkState(!safeRef.invalidated()); - checkState(safeRef.global() != null, "safeRef node is null for %s", key); - checkState(safeRef.global() == node, "safeRef node not in map: %s != %s", safeRef.global(), node); - checkState(node.references() > 0, "references (%d) are zero for %s (%s)", node.references(), key, node); - checkState(node.isUnqueued()); + require(!safeRef.invalidated()); + require(safeRef.global() != null, "safeRef node is null for %s", key); + require(safeRef.global() == node, "safeRef node not in map: %s != %s", safeRef.global(), node); + require(node.references() > 0, "references (%d) are zero for %s (%s)", node.references(), key, node); + require(node.isUnqueued()); boolean evict = false; if (safeRef.hasUpdate()) @@ -1174,7 +1174,7 @@ public class AccordCache implements CacheSize @Override public Command load(AccordCommandStore commandStore, TxnId txnId) { - Invariants.checkState(!txnId.is(Txn.Kind.EphemeralRead)); + Invariants.require(!txnId.is(Txn.Kind.EphemeralRead)); return commandStore.loadCommand(txnId); } @@ -1208,7 +1208,7 @@ public class AccordCache implements CacheSize public Object fullShrink(TxnId txnId, Command value) { if (txnId.is(Txn.Kind.EphemeralRead)) - Invariants.checkState(value.saveStatus().compareTo(SaveStatus.ReadyToExecute) < 0); + Invariants.require(value.saveStatus().compareTo(SaveStatus.ReadyToExecute) < 0); try { diff --git a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java index 4e152e1650..f03d01c558 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java @@ -84,8 +84,8 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode LOADED.permittedFrom |= 1 << MODIFIED.ordinal(); for (Status status : VALUES) { - Invariants.checkState((status.ordinal() & IS_LOADED) != 0 == status.loaded); - Invariants.checkState(((status.ordinal() & IS_LOADED) != 0 && (status.ordinal() & IS_NESTED) != 0) == status.nested); + Invariants.require((status.ordinal() & IS_LOADED) != 0 == status.loaded); + Invariants.require(((status.ordinal() & IS_LOADED) != 0 && (status.ordinal() & IS_NESTED) != 0) == status.nested); } } @@ -168,7 +168,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode boolean isNested() { - Invariants.checkState(isLoaded()); + Invariants.require(isLoaded()); return (status & IS_NESTED) != 0; } @@ -194,13 +194,13 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode int noEvictGeneration() { - Invariants.checkState(isNoEvict()); + Invariants.require(isNoEvict()); return (status >>> 8) & 0xffff; } int noEvictMaxAge() { - Invariants.checkState(isNoEvict()); + Invariants.require(isNoEvict()); return status >>> 24; } @@ -245,7 +245,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode private void setStatus(Status newStatus) { - Invariants.checkState((newStatus.permittedFrom & (1 << (status & STATUS_MASK))) != 0, "%s not permitted from %s", newStatus, status()); + Invariants.require((newStatus.permittedFrom & (1 << (status & STATUS_MASK))) != 0, "%s not permitted from %s", newStatus, status()); setStatusUnsafe(newStatus); } @@ -257,22 +257,22 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode public void initialize(V value) { - Invariants.checkState(state == null); + Invariants.require(state == null); setStatus(LOADED); state = value; } public void readyToLoad() { - Invariants.checkState(state == null); + Invariants.require(state == null); setStatus(WAITING_TO_LOAD); state = new WaitingToLoad(); } public void markNoEvict(int generation, int maxAge) { - Invariants.checkState((maxAge & ~0xff) == 0); - Invariants.checkState((generation & ~0xffff) == 0); + Invariants.require((maxAge & ~0xff) == 0); + Invariants.require((generation & ~0xffff) == 0); status |= NO_EVICT; status |= generation << 8; status |= maxAge << 24; @@ -325,7 +325,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode public

Loading load(BiFunction loadExecutor, P param, Adapter adapter, OnLoaded onLoaded) { - Invariants.checkState(is(WAITING_TO_LOAD), "%s", this); + Invariants.require(is(WAITING_TO_LOAD), "%s", this); Loading loading = ((WaitingToLoad)state).load(loadExecutor.apply(param, () -> { V result; try @@ -346,7 +346,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode public Loading testLoad() { - Invariants.checkState(is(WAITING_TO_LOAD)); + Invariants.require(is(WAITING_TO_LOAD)); Loading loading = ((WaitingToLoad)state).load(() -> {}); setStatus(LOADING); state = loading; @@ -355,7 +355,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode public Loading loading() { - Invariants.checkState(is(LOADING), "%s", this); + Invariants.require(is(LOADING), "%s", this); return (Loading) state; } @@ -363,8 +363,8 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode // but this one is less obvious so named as to draw attention public V getExclusive() { - Invariants.checkState(owner == null || owner.commandStore == null || owner.commandStore.executor().isOwningThread()); - Invariants.checkState(isLoaded(), "%s", this); + Invariants.require(owner == null || owner.commandStore == null || owner.commandStore.executor().isOwningThread()); + Invariants.require(isLoaded(), "%s", this); if (isShrunk()) { AccordCache.Type parent = owner.parent(); @@ -500,7 +500,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode protected void saved() { - Invariants.checkState(is(MODIFIED)); + Invariants.require(is(MODIFIED)); setStatus(LOADED); } @@ -525,7 +525,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode private boolean tryShrink(K key, Adapter adapter) { - Invariants.checkState(!isNested()); + Invariants.require(!isNested()); if (isShrunk() || state == null) return false; @@ -540,7 +540,7 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode private void inflate(AccordCommandStore commandStore, K key, Adapter adapter) { - Invariants.checkState(isShrunk()); + Invariants.require(isShrunk()); if (isNested()) { Nested nested = (Nested) state; diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 463812b35f..8f1e08ab71 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -75,7 +75,7 @@ import static accord.local.KeyHistory.SYNC; import static accord.primitives.Status.Committed; import static accord.primitives.Status.PreCommitted; import static accord.primitives.Status.Truncated; -import static accord.utils.Invariants.checkState; +import static accord.utils.Invariants.require; public class AccordCommandStore extends CommandStore { @@ -247,7 +247,7 @@ public class AccordCommandStore extends CommandStore public Caches cachesExclusive() { - Invariants.checkState(executor.isOwningThread()); + Invariants.require(executor.isOwningThread()); return caches; } @@ -360,14 +360,14 @@ public class AccordCommandStore extends CommandStore public AccordSafeCommandStore begin(AccordTask operation, @Nullable CommandsForRanges commandsForRanges) { - checkState(current == null); + require(current == null); current = AccordSafeCommandStore.create(operation, commandsForRanges, this); return current; } void setOwner(Thread thread, Thread self) { - Invariants.checkState(thread == null ? currentThread == self : currentThread == null); + Invariants.require(thread == null ? currentThread == self : currentThread == null); currentThread = thread; if (thread != null) CommandStore.register(this); @@ -380,7 +380,7 @@ public class AccordCommandStore extends CommandStore public void complete(AccordSafeCommandStore store) { - checkState(current == store); + require(current == store); current.postExecute(); current = null; } @@ -388,7 +388,7 @@ public class AccordCommandStore extends CommandStore public void abort(AccordSafeCommandStore store) { checkInStore(); - Invariants.checkState(store == current); + Invariants.require(store == current); current = null; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java index 62fdf894d7..4cf16f0241 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java @@ -243,7 +243,7 @@ public class AccordConfigurationService extends AbstractConfigurationService tableRange : tableRanges) @@ -156,7 +156,7 @@ public class AccordDataStore implements DataStore } else { - checkState(firstToken.compareTo(lastToken) < 0); + require(firstToken.compareTo(lastToken) < 0); org.apache.cassandra.dht.Range memtableRange = new org.apache.cassandra.dht.Range<>(firstToken, lastToken); for (org.apache.cassandra.dht.Range tableRange : tableRanges) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index 0ab6dcb2e3..626b08fe34 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -178,7 +178,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo public AccordCache cacheExclusive() { - Invariants.checkState(isOwningThread()); + Invariants.require(isOwningThread()); return cache; } @@ -270,7 +270,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo continue outer; } - Invariants.checkState(load != null); + Invariants.require(load != null); AccordCacheEntry.OnLoaded onLoaded = this; ++activeLoads; if (isForRange) @@ -286,15 +286,15 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo updateQueue(task); } Object prev = next.pollWaitingToLoad(); - Invariants.checkState(prev == load); + Invariants.require(prev == load); if (next.peekWaitingToLoad() == null) break; - Invariants.checkState(next.state() == WAITING_TO_LOAD, "Invalid state: %s", next); + Invariants.require(next.state() == WAITING_TO_LOAD, "Invalid state: %s", next); if (activeLoads >= maxQueuedLoads) return; } - Invariants.checkState(next.state().compareTo(LOADING) >= 0, "Invalid state: %s", next); + Invariants.require(next.state().compareTo(LOADING) >= 0, "Invalid state: %s", next); updateQueue(next); } } @@ -377,7 +377,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo private Cancellable submitIOExclusive(Task parent, Runnable run) { - Invariants.checkState(isOwningThread()); + Invariants.require(isOwningThread()); ++tasks; PlainRunnable task = new PlainRunnable(null, run, null); // TODO (expected): adopt queue position of the submitting task @@ -409,7 +409,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo public void cancel(AccordTask task) { - Invariants.checkState(task.commandStore.executor() == this, + Invariants.require(task.commandStore.executor() == this, "%s is a wrong command store for %s, should be %s", this, task, task); submit(AccordExecutor::cancelExclusive, CancelAsync::new, task); @@ -575,7 +575,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo { if (task.onLoad(loaded)) { - Invariants.checkState(task.queued() == loading); + Invariants.require(task.queued() == loading); task.unqueue(); waitingToRun(task); } @@ -627,14 +627,14 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo @Override public void setCapacity(long bytes) { - Invariants.checkState(isOwningThread()); + Invariants.require(isOwningThread()); cache.setCapacity(bytes); maxWorkingCapacityInBytes = cache.capacity() + maxWorkingSetSizeInBytes; } public void setWorkingSetSize(long bytes) { - Invariants.checkState(isOwningThread()); + Invariants.require(isOwningThread()); maxWorkingSetSizeInBytes = bytes; maxWorkingCapacityInBytes = cache.capacity() + maxWorkingSetSizeInBytes; if (maxWorkingCapacityInBytes < maxWorkingSetSizeInBytes) @@ -643,7 +643,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo public void setMaxQueuedLoads(int total, int range) { - Invariants.checkState(isOwningThread()); + Invariants.require(isOwningThread()); maxQueuedLoads = total; maxQueuedRangeLoads = range; } @@ -729,7 +729,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo @Override protected void preRunExclusive() { - Invariants.checkState(task != null); + Invariants.require(task != null); Thread self = Thread.currentThread(); commandStore.setOwner(self, self); task.preRunExclusive(); @@ -795,7 +795,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo @Override public void append(Task task) { - Invariants.checkState(!next.isSet()); + Invariants.require(!next.isSet()); // TODO (expected): if the new task is higher priority, replace next next.setNext(task); waitingToRun.append(next); @@ -866,9 +866,9 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo public void remove(T remove) { - Invariants.checkState(super.contains(remove)); + Invariants.require(super.contains(remove)); super.remove(remove); - Invariants.checkState(!super.contains(remove)); + Invariants.require(!super.contains(remove)); } public boolean contains(T contains) @@ -1074,7 +1074,7 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo @Override protected void addToQueue(TaskQueue queue) { - Invariants.checkState(queue.kind == WAITING_TO_RUN); + Invariants.require(queue.kind == WAITING_TO_RUN); queue.append(this); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorInfiniteLoops.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorInfiniteLoops.java index e267e0e3a8..405479bb9f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorInfiniteLoops.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorInfiniteLoops.java @@ -43,7 +43,7 @@ class AccordExecutorInfiniteLoops implements Shutdownable public AccordExecutorInfiniteLoops(Mode mode, int threads, IntFunction name, Function tasks) { - Invariants.checkState(mode == RUN_WITH_LOCK ? threads == 1 : threads >= 1); + Invariants.require(mode == RUN_WITH_LOCK ? threads == 1 : threads >= 1); final LongHashSet threadIds = new LongHashSet(threads, 0.5f); this.loops = new Interruptible[threads]; for (int i = 0; i < threads; ++i) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java index 3e8e76eb98..cd192e43c5 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java @@ -67,7 +67,7 @@ class AccordExecutorSimple extends AccordExecutor public AccordExecutorSimple(ReentrantLock lock, int executorId, Mode mode, int threads, IntFunction name, AccordCacheMetrics metrics, ExecutorFunctionFactory loadExecutor, ExecutorFunctionFactory saveExecutor, ExecutorFunctionFactory rangeLoadExecutor, Agent agent) { super(lock, executorId, metrics, loadExecutor, saveExecutor, rangeLoadExecutor, agent); - Invariants.checkArgument(threads == 1); + Invariants.requireArgument(threads == 1); this.lock = lock; this.executor = executorFactory().sequential(name.apply(0)); diff --git a/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java index de14638e77..b95e5cbde8 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java @@ -117,7 +117,7 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi public PeerStatus onUpdate(Node.Id node, Status status) { - Invariants.checkArgument(contains(node)); + Invariants.requireArgument(contains(node)); PeerStatus peerStatus = new PeerStatus(node, status); statusMap.put(node, peerStatus); return peerStatus; diff --git a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java index 9d91e386e8..f56c388012 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java @@ -155,7 +155,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements if (that.streams.keySet().stream().anyMatch(this.streams::containsKey)) throw new IllegalStateException(String.format("Unable to merge: key found in multiple StreamData %s %s", this.streams.keySet(), that.streams.keySet())); - Invariants.checkState(!that.streams.keySet().stream().anyMatch(this.streams::containsKey)); + Invariants.require(!that.streams.keySet().stream().anyMatch(this.streams::containsKey)); ImmutableMap.Builder builder = ImmutableMap.builder(); builder.putAll(this.streams); builder.putAll(that.streams); @@ -180,8 +180,8 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements { Invariants.nonNull(range); Invariants.nonNull(from); - Invariants.checkState(this.range == null, "range was not null: %s", this.range); - Invariants.checkState(this.from == null, "from was not null: %s", this.from); + Invariants.require(this.range == null, "range was not null: %s", this.range); + Invariants.require(this.from == null, "from was not null: %s", this.from); this.range = range; this.from = from; maybeListen(); @@ -190,7 +190,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements private void futureReceived(StreamResultFuture future) { Invariants.nonNull(future); - Invariants.checkState(this.future == null, "future was not null: %s", this.future); + Invariants.require(this.future == null, "future was not null: %s", this.future); this.future = future; maybeListen(); } @@ -254,7 +254,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements if (!session.peer.equals(to)) continue; - Invariants.checkState(session.getNumRequests() == 0, "Requested to send data: %s", session); + Invariants.require(session.getNumRequests() == 0, "Requested to send data: %s", session); if (session.getNumTransfers() > 0) return true; } @@ -266,14 +266,14 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements { try { - Invariants.checkArgument(key.domain() == Routable.Domain.Range, "Required Range but saw %s: %s", key.domain(), key); + Invariants.requireArgument(key.domain() == Routable.Domain.Range, "Required Range but saw %s: %s", key.domain(), key); TokenRange range = (TokenRange) key; // TODO (required): check epoch // TODO (required): handle dropped tables TableId tableId = range.table(); TableMetadata table = ClusterMetadata.current().schema.getKeyspaces().getTableOrViewNullable(tableId); - Invariants.checkState(table != null, "Table with id %s not found", tableId); + Invariants.require(table != null, "Table with id %s not found", tableId); // TODO (required): may also be relocation StreamPlan plan = new StreamPlan(StreamOperation.BOOTSTRAP, 1, false, @@ -310,7 +310,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements @Override public void serialize(Query t, DataOutputPlus out, int version) { - Invariants.checkArgument(t == noopQuery); + Invariants.requireArgument(t == noopQuery); } @Override @@ -322,7 +322,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements @Override public long serializedSize(Query t, int version) { - Invariants.checkArgument(t == noopQuery); + Invariants.requireArgument(t == noopQuery); return 0; } }; @@ -332,7 +332,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements @Override public void serialize(Update t, DataOutputPlus out, int version) { - Invariants.checkArgument(t == null); + Invariants.requireArgument(t == null); } @Override @@ -344,7 +344,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements @Override public long serializedSize(Update t, int version) { - Invariants.checkArgument(t == null); + Invariants.requireArgument(t == null); return 0; } }; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 1776c76b2d..9be8739416 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -98,7 +98,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier static { // make noise early if we forget to update our version mappings - Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_51, "Expected current version to be %d but given %d", MessagingService.VERSION_51, MessagingService.current_version); + Invariants.require(MessagingService.current_version == MessagingService.VERSION_51, "Expected current version to be %d but given %d", MessagingService.VERSION_51, MessagingService.current_version); } static final ThreadLocal keyCRCBytes = ThreadLocal.withInitial(() -> new byte[JournalKeySupport.TOTAL_SIZE]); @@ -162,7 +162,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public AccordJournal start(Node node) { - Invariants.checkState(status == Status.INITIALIZED); + Invariants.require(status == Status.INITIALIZED); this.node = node; status = Status.STARTING; journal.start(); @@ -193,7 +193,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @Override public void shutdown() { - Invariants.checkState(status == Status.REPLAY || status == Status.STARTED, "%s", status); + Invariants.require(status == Status.REPLAY || status == Status.STARTED, "%s", status); status = Status.TERMINATING; journal.shutdown(); status = Status.TERMINATED; @@ -251,7 +251,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier case ERASE: return null; } - Invariants.checkState(builder.saveStatus() != null, "No saveSatus loaded, but next was called and cleanup was not: %s", builder); + Invariants.require(builder.saveStatus() != null, "No saveSatus loaded, but next was called and cleanup was not: %s", builder); return builder.asMinimal(); } @@ -411,7 +411,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier builder.forceResult(orig.result()); // We can only use strict equality if we supply result. Command reconstructed = builder.construct(redundantBefore); - Invariants.checkState(orig.equals(reconstructed), + Invariants.require(orig.equals(reconstructed), '\n' + "Original: %s\n" + "Reconstructed: %s\n" + @@ -460,7 +460,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier JournalKey finalKey = key; iter.readAllForKey(key, (segment, position, local, buffer, userVersion) -> { - Invariants.checkState(finalKey.equals(local)); + Invariants.require(finalKey.equals(local)); try (DataInputBuffer in = new DataInputBuffer(buffer, false)) { builder.deserializeNext(in, userVersion); @@ -481,7 +481,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier { CommandStore commandStore = commandStores.forId(key.commandStoreId); Command command = builder.construct(commandStore.unsafeGetRedundantBefore()); - Invariants.checkState(command.saveStatus() != SaveStatus.Uninitialised, + Invariants.require(command.saveStatus() != SaveStatus.Uninitialised, "Found uninitialized command in the log: %s %s", command.toString(), builder.toString()); Loader loader = commandStore.loader(); async(loader::load, command).get(); @@ -572,7 +572,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier private static void serialize(Command command, int flags, DataOutputPlus out, int userVersion) throws IOException { - Invariants.checkState(flags != 0); + Invariants.require(flags != 0); out.writeInt(flags); int iterable = toIterableSetFields(flags); @@ -679,8 +679,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public void serialize(DataOutputPlus out, RedundantBefore redundantBefore, int userVersion) throws IOException { - Invariants.checkState(mask == 0); - Invariants.checkState(flags != 0); + Invariants.require(mask == 0); + Invariants.require(flags != 0); int flags = validateFlags(this.flags); Writer.serialize(construct(redundantBefore), flags, out, userVersion); @@ -688,9 +688,9 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public void deserializeNext(DataInputPlus in, int userVersion) throws IOException { - Invariants.checkState(txnId != null); + Invariants.require(txnId != null); int flags = in.readInt(); - Invariants.checkState(flags != 0); + Invariants.require(flags != 0); nextCalled = true; count++; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java index 9f83eba75f..95267a6b79 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java @@ -325,7 +325,7 @@ public class AccordJournalTable implements RangeSearche { UnfilteredRowIterator next = partitionIterator.next(); JournalKey partitionKeyComponents = AccordKeyspace.JournalColumns.getJournalKey(next.partitionKey()); - Invariants.checkState(partitionKeyComponents.commandStoreId == storeId, + Invariants.require(partitionKeyComponents.commandStoreId == storeId, () -> String.format("table index returned a command store other than the exepcted one; expected %d != %d", storeId, partitionKeyComponents.commandStoreId)); return partitionKeyComponents.id; } @@ -372,7 +372,7 @@ public class AccordJournalTable implements RangeSearche private void readRow(K key, Unfiltered unfiltered, EntryHolder into, RecordConsumer onEntry) { - Invariants.checkState(unfiltered.isRow()); + Invariants.require(unfiltered.isRow()); Row row = (Row) unfiltered; long descriptor = LongType.instance.compose(ByteBuffer.wrap((byte[]) row.clustering().get(0))); diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java index eb60d901df..9777f352bd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java @@ -335,7 +335,7 @@ public class AccordJournalValueSerializers epochs[i] = in.readLong(); ranges[i] = KeySerializers.ranges.deserialize(in, messagingVersion); } - Invariants.checkState(ranges.length == epochs.length); + Invariants.require(ranges.length == epochs.length); into.update(new RangesForEpoch(epochs, ranges)); } } @@ -354,7 +354,7 @@ public class AccordJournalValueSerializers protected NavigableMap accumulate(NavigableMap accumulator, V newValue) { V prev = accumulator.put(getKey.apply(newValue), newValue); - Invariants.checkState(prev == null || prev.equals(newValue)); + Invariants.require(prev == null || prev.equals(newValue)); return accumulator; } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index 57aab6653a..1f46e406b5 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -131,7 +131,7 @@ import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.btree.BTreeSet; import org.apache.cassandra.utils.concurrent.OpOrder; -import static accord.utils.Invariants.checkState; +import static accord.utils.Invariants.require; import static java.lang.String.format; import static java.util.Collections.emptyMap; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; @@ -327,7 +327,7 @@ public class AccordKeyspace // TODO (expected): garbage-free filtering, reusing encoding public Row withoutRedundantCommands(TokenKey key, Row row, RedundantBefore.Entry redundantBefore) { - Invariants.checkState(row.columnCount() == 1); + Invariants.require(row.columnCount() == 1); Cell cell = row.getCell(data); if (cell == null) return row; @@ -688,7 +688,7 @@ public class AccordKeyspace try (RowIterator partition = partitions.next()) { - Invariants.checkState(partition.hasNext()); + Invariants.require(partition.hasNext()); Row row = partition.next(); ByteBuffer data = cellValue(row, accessor.data); return Serialize.fromBytes(key, data); @@ -714,8 +714,8 @@ public class AccordKeyspace private EpochDiskState(long minEpoch, long maxEpoch) { - Invariants.checkArgument(minEpoch >= 0, "Min Epoch %d < 0", minEpoch); - Invariants.checkArgument(maxEpoch >= minEpoch, "Max epoch %d < min %d", maxEpoch, minEpoch); + Invariants.requireArgument(minEpoch >= 0, "Min Epoch %d < 0", minEpoch); + Invariants.requireArgument(maxEpoch >= minEpoch, "Max epoch %d < min %d", maxEpoch, minEpoch); this.minEpoch = minEpoch; this.maxEpoch = maxEpoch; } @@ -740,14 +740,14 @@ public class AccordKeyspace @VisibleForTesting EpochDiskState withNewMaxEpoch(long epoch) { - Invariants.checkArgument(epoch > maxEpoch, "Epoch %d <= %d (max)", epoch, maxEpoch); + Invariants.requireArgument(epoch > maxEpoch, "Epoch %d <= %d (max)", epoch, maxEpoch); return EpochDiskState.create(Math.max(1, minEpoch), epoch); } private EpochDiskState withNewMinEpoch(long epoch) { - Invariants.checkArgument(epoch > minEpoch, "epoch %d <= %d (min)", epoch, minEpoch); - Invariants.checkArgument(epoch <= maxEpoch, "epoch %d > %d (max)", epoch, maxEpoch); + Invariants.requireArgument(epoch > minEpoch, "epoch %d <= %d (min)", epoch, minEpoch); + Invariants.requireArgument(epoch <= maxEpoch, "epoch %d > %d (max)", epoch, maxEpoch); return EpochDiskState.create(epoch, maxEpoch); } @@ -792,7 +792,7 @@ public class AccordKeyspace CommandSerializers.txnId.serialize(key.id, id); id.flip(); ByteBuffer pk = keyComparator.make(key.commandStoreId, (byte)key.type.id, id).serializeAsPartitionKey(); - Invariants.checkState(getTxnId(splitPartitionKey(pk)).equals(key.id)); + Invariants.require(getTxnId(splitPartitionKey(pk)).equals(key.id)); return Journal.partitioner.decorateKey(pk); } @@ -871,7 +871,7 @@ public class AccordKeyspace { if (diskState.isEmpty()) return saveEpochDiskState(EpochDiskState.create(epoch)); - Invariants.checkArgument(epoch >= diskState.minEpoch, "Epoch %d < %d (min)", epoch, diskState.minEpoch); + Invariants.requireArgument(epoch >= diskState.minEpoch, "Epoch %d < %d (min)", epoch, diskState.minEpoch); if (epoch > diskState.maxEpoch) { diskState = diskState.withNewMaxEpoch(epoch); @@ -900,20 +900,16 @@ public class AccordKeyspace diskState = maybeUpdateMaxEpoch(diskState, epoch); String cql = "UPDATE " + ACCORD_KEYSPACE_NAME + '.' + TOPOLOGIES + ' ' + "SET closed = closed + ? WHERE epoch = ?"; - executeInternal(cql, - KeySerializers.rangesToBlobMap(ranges), epoch); + executeInternal(cql, KeySerializers.rangesToBlobMap(ranges), epoch); return diskState; } - // TODO (required): unused public static EpochDiskState markRetired(Ranges ranges, long epoch, EpochDiskState diskState) { diskState = maybeUpdateMaxEpoch(diskState, epoch); String cql = "UPDATE " + ACCORD_KEYSPACE_NAME + '.' + TOPOLOGIES + ' ' + "SET retired = retired + ? WHERE epoch = ?"; - executeInternal(cql, - KeySerializers.rangesToBlobMap(ranges), epoch); - flush(Topologies); + executeInternal(cql, KeySerializers.rangesToBlobMap(ranges), epoch); return diskState; } @@ -979,7 +975,7 @@ public class AccordKeyspace consumer.load(epoch, SyncStatus.NOT_STARTED, Collections.emptySet(), Collections.emptySet(), Ranges.EMPTY, Ranges.EMPTY); return; } - checkState(!result.isEmpty(), "Nothing found for epoch %d", epoch); + require(!result.isEmpty(), "Nothing found for epoch %d", epoch); UntypedResultSet.Row row = result.one(); SyncStatus syncStatus = row.has("sync_state") diff --git a/src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java b/src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java index c581f59d34..995ba7f9e9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java @@ -59,7 +59,7 @@ public class AccordSegmentCompactor implements SegmentCompactor> compact(Collection> segments) { - Invariants.checkState(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size())); + Invariants.require(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size())); logger.info("Compacting {} static segments: {}", segments.size(), segments); PriorityQueue> readers = new PriorityQueue<>(); @@ -110,9 +110,9 @@ public class AccordSegmentCompactor implements SegmentCompactor implements SegmentCompactor= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1), + Invariants.requireArgument((decoratedKey.compareTo(prevDecoratedKey) >= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1), String.format("Partition key and JournalKey didn't have matching order, which may imply a serialization issue.\n%s (%s)\n%s (%s)", key, decoratedKey, prevKey, prevDecoratedKey)); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index ca98fa7a0e..0c89bcb734 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -177,7 +177,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static accord.messages.SimpleReply.Ok; import static accord.primitives.Routable.Domain.Key; import static accord.primitives.Routable.Domain.Range; -import static accord.utils.Invariants.checkState; +import static accord.utils.Invariants.require; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.concurrent.TimeUnit.SECONDS; @@ -310,7 +310,7 @@ public class AccordService implements IAccordService, Shutdownable if (!DatabaseDescriptor.getAccordTransactionsEnabled()) return NOOP_SERVICE; IAccordService i = instance; - Invariants.checkState(i != null, "AccordService was not started"); + Invariants.require(i != null, "AccordService was not started"); return i; } @@ -324,7 +324,7 @@ public class AccordService implements IAccordService, Shutdownable @VisibleForTesting public AccordService(Id localId) { - Invariants.checkState(localId != null, "static localId must be set before instantiating AccordService"); + Invariants.require(localId != null, "static localId must be set before instantiating AccordService"); logger.info("Starting accord with nodeId {}", localId); AccordAgent agent = FBUtilities.construct(CassandraRelevantProperties.ACCORD_AGENT_CLASS.getString(AccordAgent.class.getName()), "AccordAgent"); agent.setNodeId(localId); @@ -702,7 +702,7 @@ public class AccordService implements IAccordService, Shutdownable if (success == null) { logger.error("Ran out of retries for barrier"); - checkState(existingFailures != null, "Didn't have success, but also didn't have failures"); + require(existingFailures != null, "Didn't have success, but also didn't have failures"); Throwables.throwIfUnchecked(existingFailures); throw new RuntimeException(existingFailures); } @@ -1260,7 +1260,7 @@ public class AccordService implements IAccordService, Shutdownable if (ranges.isEmpty()) return; // nothing to see here ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(id); - Invariants.checkState(cfs != null, "Unable to find table %s", id); + Invariants.require(cfs != null, "Unable to find table %s", id); BigInteger targetSplitSize = BigInteger.valueOf(Math.max(1, cfs.estimateKeys() / 1_000_000)); List> syncs = new ArrayList<>(ranges.size()); diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java index b7f74257cb..5e2e0b8a85 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java @@ -316,7 +316,7 @@ public class AccordSyncPropagator @Override public void onResponse(Message msg) { - Invariants.checkState(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg); + Invariants.require(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg); Set completedEpochs = new HashSet<>(); // TODO review is it a good idea to call the listener while not holding the `AccordSyncPropagator` lock? synchronized (AccordSyncPropagator.this) diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index 1f258fcf50..131950ced8 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -287,12 +287,12 @@ public abstract class AccordTask extends Task implements Runnable, Function no loading or waiting; found %s", this, AccordTask::toDescription); + Invariants.require(rangeScanner == null || rangeScanner.scanned); + Invariants.require(loading == null && waitingToLoad == null, "WAITING_TO_RUN => no loading or waiting; found %s", this, AccordTask::toDescription); loadedAt = nanoTime(); } else if (state == RUNNING) @@ -317,7 +317,7 @@ public abstract class AccordTask extends Task implements Runnable, Function callback) { - Invariants.checkState(AccordTask.this.callback == null); + Invariants.require(AccordTask.this.callback == null); AccordTask.this.callback = callback; commandStore.tryPreSetup(AccordTask.this); commandStore.executor().submit(AccordTask.this); @@ -396,7 +396,7 @@ public abstract class AccordTask extends Task implements Runnable, Function extends Task implements Runnable, Function node = ref.global(); int refs = node.increment(); - Invariants.checkState(refs > 1); + Invariants.require(refs > 1); loaded.apply(this).put(k, cache.parent().adapter().safeRef(node)); } @@ -498,7 +498,7 @@ public abstract class AccordTask extends Task implements Runnable, Function state) { AccordSafeState safeRef = loading == null ? null : loading.remove(state.key()); - Invariants.checkState(safeRef != null && safeRef.global() == state, "Expected to find %s loading; found %s", state, this, AccordTask::toDescription); + Invariants.require(safeRef != null && safeRef.global() == state, "Expected to find %s loading; found %s", state, this, AccordTask::toDescription); if (safeRef.getClass() == AccordSafeCommand.class) ensureCommands().put((TxnId)state.key(), (AccordSafeCommand) safeRef); else @@ -511,7 +511,7 @@ public abstract class AccordTask extends Task implements Runnable, Function extends Task implements Runnable, Function state) { boolean removed = waitingToLoad != null && waitingToLoad.remove(state); - Invariants.checkState(removed, "%s not found in waitingToLoad %s", state, this, AccordTask::toDescription); + Invariants.require(removed, "%s not found in waitingToLoad %s", state, this, AccordTask::toDescription); if (!waitingToLoad.isEmpty()) return false; @@ -575,7 +575,7 @@ public abstract class AccordTask extends Task implements Runnable, Function> ensureWaitingToLoad() { - Invariants.checkState(state.compareTo(WAITING_TO_LOAD) <= 0, "Expected status to be on or before WAITING_TO_LOAD; found %s", this, AccordTask::toDescription); + Invariants.require(state.compareTo(WAITING_TO_LOAD) <= 0, "Expected status to be on or before WAITING_TO_LOAD; found %s", this, AccordTask::toDescription); if (waitingToLoad == null) waitingToLoad = new ArrayDeque<>(); return waitingToLoad; @@ -583,7 +583,7 @@ public abstract class AccordTask extends Task implements Runnable, Function pollWaitingToLoad() { - Invariants.checkState(state == State.WAITING_TO_LOAD, "Expected status to be WAITING_TO_LOAD; found %s", this, AccordTask::toDescription); + Invariants.require(state == State.WAITING_TO_LOAD, "Expected status to be WAITING_TO_LOAD; found %s", this, AccordTask::toDescription); if (waitingToLoad == null) return null; @@ -612,7 +612,7 @@ public abstract class AccordTask extends Task implements Runnable, Function extends Task implements Runnable, Function extends Task implements Runnable, Function range : ranges) - Invariants.checkArgument(!range.isWrapAround() || range.right.equals(range.right.minValue()), + Invariants.requireArgument(!range.isWrapAround() || range.right.equals(range.right.minValue()), "wrap around range %s found", range); Sets.SetView readOnly = Sets.difference(readEndpoints, writeEndpoints); - Invariants.checkState(readOnly.isEmpty(), "Read only replicas detected: %s", readOnly); + Invariants.require(readOnly.isEmpty(), "Read only replicas detected: %s", readOnly); SortedArrayList nodes = new SortedArrayList<>(writeEndpoints.stream() .map(directory::peerId) diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index 14326aefb3..a417e36198 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -227,7 +227,7 @@ public class CommandsForRanges extends TreeMap implements Co public Summary ifRelevant(Command.Minimal cmd) { - Invariants.checkState(findAsDep == null); + Invariants.require(findAsDep == null); return ifRelevant(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null); } } diff --git a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java index 5745c55112..78ec8c4589 100644 --- a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java +++ b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java @@ -85,16 +85,16 @@ class EndpointMapping implements AccordEndpointMapper public Builder add(InetAddressAndPort endpoint, Node.Id id) { - Invariants.checkArgument(!mapping.containsKey(id), "Mapping already exists for Node.Id %s", id); - Invariants.checkArgument(!mapping.containsValue(endpoint), "Mapping already exists for %s", endpoint); + Invariants.requireArgument(!mapping.containsKey(id), "Mapping already exists for Node.Id %s", id); + Invariants.requireArgument(!mapping.containsValue(endpoint), "Mapping already exists for %s", endpoint); mapping.put(id, endpoint); return this; } public Builder removed(InetAddressAndPort endpoint, Node.Id id, long epoch) { - Invariants.checkArgument(!mapping.containsKey(id), "Mapping already exists for Node.Id %s", id); - Invariants.checkArgument(!mapping.containsValue(endpoint), "Mapping already exists for %s", endpoint); + Invariants.requireArgument(!mapping.containsKey(id), "Mapping already exists for Node.Id %s", id); + Invariants.requireArgument(!mapping.containsValue(endpoint), "Mapping already exists for %s", endpoint); mapping.put(id, endpoint); removed.put(id, epoch); return this; diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java index f11154ff83..8baa85be2e 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -52,8 +52,8 @@ public final class JournalKey public JournalKey(TxnId id, Type type, int commandStoreId) { - Invariants.checkArgument(commandStoreId >= 0); - Invariants.checkState((id.lsb & (0xffff & ~TxnId.IDENTITY_FLAGS)) == 0); + Invariants.requireArgument(commandStoreId >= 0); + Invariants.require((id.lsb & (0xffff & ~TxnId.IDENTITY_FLAGS)) == 0); Invariants.nonNull(type); Invariants.nonNull(id); this.type = type; diff --git a/src/java/org/apache/cassandra/service/accord/TokenRange.java b/src/java/org/apache/cassandra/service/accord/TokenRange.java index 6db7696330..1f684fb4b8 100644 --- a/src/java/org/apache/cassandra/service/accord/TokenRange.java +++ b/src/java/org/apache/cassandra/service/accord/TokenRange.java @@ -44,7 +44,7 @@ public class TokenRange extends Range.EndInclusive public static TokenRange create(AccordRoutingKey start, AccordRoutingKey end) { - Invariants.checkArgument(start.table().equals(end.table()), + Invariants.requireArgument(start.table().equals(end.table()), "Token ranges cannot cover more than one keyspace start:%s, end:%s", start, end); return new TokenRange(start, end); diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index a44c76cc66..c5561be97c 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -204,7 +204,7 @@ public class AccordAgent implements Agent startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), oneSecond, random); long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis()); long delayMicros = Math.max(1, startTime - nowMicros); - Invariants.checkState(delayMicros < TimeUnit.HOURS.toMicros(1L)); + Invariants.require(delayMicros < TimeUnit.HOURS.toMicros(1L)); return units.convert(delayMicros, MICROSECONDS); } diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java index bbc938516c..8aef6abe6a 100644 --- a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java +++ b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java @@ -199,7 +199,7 @@ public class ParameterizedFastPathStrategy implements FastPathStrategy cmp = this.sortPos - that.sortPos; if (cmp != 0) return cmp; - Invariants.checkState(this.id.equals(that.id)); + Invariants.require(this.id.equals(that.id)); return 0; } } @@ -228,7 +228,7 @@ public class ParameterizedFastPathStrategy implements FastPathStrategy Arrays.sort(array); SortedArrayList electorate = new SortedArrayList<>(array); - Invariants.checkState(electorate.size() >= slowQuorum); + Invariants.require(electorate.size() >= slowQuorum); return electorate; } diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java index c69f0d840e..9540fb84fa 100644 --- a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java +++ b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java @@ -59,7 +59,7 @@ public class SimpleFastPathStrategy implements FastPathStrategy Node.Id[] array = new Node.Id[nodes.size() - discarded]; System.arraycopy(tmp, 0, array, 0, nodes.size() - discarded); SortedArrayList fastPath = new SortedArrayList<>(array); - Invariants.checkState(fastPath.size() >= Shard.slowQuorumSize(nodes.size())); + Invariants.require(fastPath.size() >= Shard.slowQuorumSize(nodes.size())); return fastPath; } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java index 8f4e51116a..0f7924179b 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java @@ -49,7 +49,7 @@ import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType; import org.apache.cassandra.service.accord.serializers.ApplySerializers.ApplySerializer; import org.apache.cassandra.service.accord.txn.AccordUpdate; -import static accord.utils.Invariants.checkState; +import static accord.utils.Invariants.requireArgument; import static com.google.common.base.Preconditions.checkArgument; /** @@ -101,7 +101,7 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL public ApplyReply apply(SafeCommandStore safeStore, StoreParticipants participants) { ApplyReply reply = super.apply(safeStore, participants); - checkState(reply == ApplyReply.Redundant || reply == ApplyReply.Applied || reply == ApplyReply.Insufficient, "Unexpected ApplyReply"); + requireArgument(reply == ApplyReply.Redundant || reply == ApplyReply.Applied || reply == ApplyReply.Insufficient, "Unexpected ApplyReply"); // Hasn't necessarily finished applying yet so need to check and maybe add a listener // Redundant means we are competing with a recovery coordinator which is fine diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index 8330f30f2c..0a6a10f5ae 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -99,7 +99,7 @@ import org.apache.cassandra.transport.Dispatcher; import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard; import static accord.primitives.Txn.Kind.Write; -import static accord.utils.Invariants.checkArgument; +import static accord.utils.Invariants.requireArgument; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteMetrics; @@ -172,7 +172,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute route, Participants readScope, Timestamp executeAt, Deps deps, BiConsumer callback, AgentExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) { - checkArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR); + requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR); this.node = node; this.txnId = txnId; this.txn = txn; @@ -183,7 +183,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen this.callback = callback; this.executor = executor; - checkArgument(updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR || consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL); + requireArgument(updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR || consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL); this.consistencyLevel = consistencyLevel; this.endpointMapper = endpointMapper; @@ -242,7 +242,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen @Override public void sendReadRepairMutation(Message message, InetAddressAndPort to, RequestCallback callback) { - checkArgument(message.payload.allowsPotentialTransactionConflicts()); + requireArgument(message.payload.allowsPotentialTransactionConflicts()); Node.Id id = endpointMapper.mappedId(to); AccordInteropReadRepair readRepair = new AccordInteropReadRepair(id, executes, txnId, readScope, executeAt.epoch(), message.payload); node.send(id, readRepair, executor, new AccordInteropReadRepair.ReadRepairCallback(id, to, message, callback, this)); diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java index 08d09f8960..1a5d74b28d 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java @@ -111,14 +111,14 @@ public class AccordInteropPersist extends Persist public AccordInteropPersist(Node node, Topologies topologies, TxnId txnId, Route sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute, ConsistencyLevel consistencyLevel, BiConsumer clientCallback) { super(node, topologies, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute); - Invariants.checkArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE); + Invariants.requireArgument(consistencyLevel == ConsistencyLevel.QUORUM || consistencyLevel == ConsistencyLevel.ALL || consistencyLevel == ConsistencyLevel.SERIAL || consistencyLevel == ConsistencyLevel.ONE); this.consistencyLevel = consistencyLevel; registerClientCallback(result, clientCallback); } public void registerClientCallback(Result result, BiConsumer clientCallback) { - Invariants.checkState(callback == null); + Invariants.require(callback == null); switch (consistencyLevel) { case ONE: // Can safely upgrade ONE to QUORUM/SERIAL to get a synchronous commit diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java index b2a9cc17b4..908d1f3cf7 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java @@ -60,7 +60,7 @@ public abstract class AccordInteropReadCallback implements Callback= -1); + Invariants.require(await.callbackId >= -1); } @Override @@ -62,7 +62,7 @@ public class AwaitSerializer long maxAwaitEpoch = in.readUnsignedVInt() + txnId.epoch(); long minAwaitEpoch = maxAwaitEpoch - in.readUnsignedVInt(); int callbackId = in.readUnsignedVInt32() - 1; - Invariants.checkState(callbackId >= -1); + Invariants.require(callbackId >= -1); return Await.SerializerSupport.create(txnId, scope, blockedUntil, notifyProgressLog, minAwaitEpoch, maxAwaitEpoch, callbackId); } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java index 436849e290..227e5c74f3 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -208,7 +208,7 @@ public class CommandSerializers { if ((flags & 1) != 0) return; - flags >>= 1; + flags >>>= 1; } in.readUnsignedVInt(); in.readUnsignedVInt(); @@ -233,14 +233,14 @@ public class CommandSerializers out.writeUnsignedVInt32(flags); if (executeAt == null) { - Invariants.checkState(nullable); + Invariants.require(nullable); } else { out.writeUnsignedVInt(executeAt.epoch()); out.writeUnsignedVInt(executeAt.hlc()); out.writeUnsignedVInt32(executeAt.node.id); - if ((flags & HAS_UNIQUE_HLC) != 0) + if (executeAt.hasDistinctHlcAndUniqueHlc()) out.writeUnsignedVInt(executeAt.uniqueHlc() - executeAt.hlc()); } } @@ -261,13 +261,13 @@ public class CommandSerializers long size = TypeSizes.sizeofUnsignedVInt(flags); if (executeAt == null) { - Invariants.checkState(nullable); + Invariants.require(nullable); return size; } size += TypeSizes.sizeofUnsignedVInt(executeAt.epoch()); size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc()); size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id); - if ((flags & HAS_UNIQUE_HLC) != 0) + if (executeAt.hasDistinctHlcAndUniqueHlc()) size += TypeSizes.sizeofUnsignedVInt(executeAt.uniqueHlc() - executeAt.hlc()); return size; } @@ -276,7 +276,7 @@ public class CommandSerializers { if (executeAt == null) { - Invariants.checkState(nullable); + Invariants.require(nullable); return 1; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java index 10433feaf4..1a1a767176 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java @@ -138,7 +138,7 @@ public class CommandStoreSerializers public void serialize(RedundantBefore.Entry t, DataOutputPlus out, int version) throws IOException { KeySerializers.range.serialize(t.range, out, version); - Invariants.checkState(t.startOwnershipEpoch <= t.endOwnershipEpoch); + Invariants.require(t.startOwnershipEpoch <= t.endOwnershipEpoch); out.writeUnsignedVInt(t.startOwnershipEpoch); if (t.endOwnershipEpoch == Long.MAX_VALUE) out.writeUnsignedVInt(0L); else out.writeUnsignedVInt(1 + t.endOwnershipEpoch - t.startOwnershipEpoch); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java index 2182d359a2..7ca3b4710e 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java @@ -39,7 +39,7 @@ public class SmallEnumSerializer> implements IVersionedSeriali public SmallEnumSerializer(Class clazz) { this.values = clazz.getEnumConstants(); - Invariants.checkArgument(values.length < 255); // allow an extra 1 for nullable variant to ensure consistency + Invariants.requireArgument(values.length < 255); // allow an extra 1 for nullable variant to ensure consistency } public E forOrdinal(int ordinal) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java index a538e4b756..528a0acb45 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java @@ -41,7 +41,7 @@ public class WaitingOnSerializer { public static void serializeBitSetsOnly(TxnId txnId, WaitingOn waitingOn, DataOutputPlus out) throws IOException { - Invariants.checkState(txnId.is(Key) == (waitingOn.appliedOrInvalidated == null)); + Invariants.require(txnId.is(Key) == (waitingOn.appliedOrInvalidated == null)); int keyCount = waitingOn.keys.size(); int txnIdCount = waitingOn.txnIdCount(); int waitingOnLength = (txnIdCount + keyCount + 63) / 64; @@ -78,8 +78,8 @@ public class WaitingOnSerializer RangeDeps directRangeDeps = deps.rangeDeps; KeyDeps directKeyDeps = deps.directKeyDeps; int txnIdCount = directRangeDeps.txnIdCount() + directKeyDeps.txnIdCount(); - Invariants.checkState(waitingOn.size()/64 == (txnIdCount + keys.size() + 63) / 64); - Invariants.checkState(appliedOrInvalidated == null || (appliedOrInvalidated.size()/64 == (txnIdCount + 63)/64)); + Invariants.require(waitingOn.size()/64 == (txnIdCount + keys.size() + 63) / 64); + Invariants.require(appliedOrInvalidated == null || (appliedOrInvalidated.size()/64 == (txnIdCount + 63)/64)); WaitingOn result = new WaitingOn(keys, directRangeDeps, directKeyDeps, waitingOn, appliedOrInvalidated); if (executeAtLeast != null) return new Command.WaitingOnWithExecuteAt(result, executeAtLeast); @@ -102,7 +102,7 @@ public class WaitingOnSerializer private static void serialize(int length, SimpleBitSet write, DataOutputPlus out) throws IOException { long[] bits = SimpleBitSet.SerializationSupport.getArray(write); - Invariants.checkState(length == bits.length); + Invariants.require(length == bits.length); for (int i = 0; i < length; i++) out.writeLong(bits[i]); } @@ -118,7 +118,7 @@ public class WaitingOnSerializer public static long serializedSize(int length, SimpleBitSet write) { long[] bits = SimpleBitSet.SerializationSupport.getArray(write); - Invariants.checkState(length == bits.length, "Expected length %d != %d", length, bits.length); + Invariants.require(length == bits.length, "Expected length %d != %d", length, bits.length); return (long) TypeSizes.LONG_SIZE * length; } } diff --git a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java index 24d3ef78ab..1bf46889dd 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java +++ b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdateParameters.java @@ -26,9 +26,13 @@ import com.google.common.collect.ImmutableMap; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.UpdateParameters; import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.TimeUUIDType; import org.apache.cassandra.db.partitions.Partition; +import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.paxos.Ballot; +import org.apache.cassandra.utils.TimeUUID; import static com.google.common.base.Preconditions.checkState; import static java.util.concurrent.TimeUnit.MICROSECONDS; @@ -46,6 +50,22 @@ public class AccordUpdateParameters this.timestamp = timestamp; } + static class RowUpdateParameters extends UpdateParameters + { + private long timeUuidNanos; + + public RowUpdateParameters(TableMetadata metadata, ClientState clientState, QueryOptions options, long timestamp, long nowInSec, int ttl, Map prefetchedRows) throws InvalidRequestException + { + super(metadata, clientState, options, timestamp, nowInSec, ttl, prefetchedRows); + } + + @Override + public byte[] nextTimeUUIDAsBytes() + { + return TimeUUID.toBytes(Ballot.unixMicrosToMsb(timestamp), TimeUUIDType.signedBytesToNativeLong(timeUuidNanos++)); + } + } + public TxnData getData() { return data; @@ -60,13 +80,13 @@ public class AccordUpdateParameters // TODO : How should Accord work with TTL? int ttl = metadata.params.defaultTimeToLive; - return new UpdateParameters(metadata, - disabledGuardrails, - options, - timestamp, - MICROSECONDS.toSeconds(timestamp), - ttl, - prefetchRow(metadata, dk, rowIndex)); + return new RowUpdateParameters(metadata, + disabledGuardrails, + options, + timestamp, + MICROSECONDS.toSeconds(timestamp), + ttl, + prefetchRow(metadata, dk, rowIndex)); } private Map prefetchRow(TableMetadata metadata, DecoratedKey dk, int index) diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnData.java b/src/java/org/apache/cassandra/service/accord/txn/TxnData.java index 8eed778a99..80920faa89 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnData.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnData.java @@ -35,7 +35,7 @@ import org.apache.cassandra.utils.Int32Serializer; import org.apache.cassandra.utils.NullableSerializer; import org.apache.cassandra.utils.ObjectSizes; -import static accord.utils.Invariants.checkArgument; +import static accord.utils.Invariants.requireArgument; import static org.apache.cassandra.service.accord.txn.TxnResult.Kind.txn_data; /** @@ -84,7 +84,7 @@ public class TxnData extends Int2ObjectHashMap implements TxnResul public static int txnDataName(TxnDataNameKind kind, int index) { - checkArgument(index >= 0 && index <= TXN_DATA_NAME_INDEX_MAX); + requireArgument(index >= 0 && index <= TXN_DATA_NAME_INDEX_MAX); int kindInt = (int)(((long)kind.value) << TXN_DATA_NAME_INDEX_BITS); return kindInt | index; } @@ -121,7 +121,7 @@ public class TxnData extends Int2ObjectHashMap implements TxnResul public static TxnData newWithExpectedSize(int size) { - checkArgument(size >= 0, "size can't be negative"); + requireArgument(size >= 0, "size can't be negative"); size = Math.max(4, size); return new TxnData(size < 1073741824 ? (int)((float)size / 0.75F + 1.0F) : Integer.MAX_VALUE); } diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index c72d52c377..d2fb6aa307 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -53,7 +53,7 @@ import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.ObjectSizes; -import static accord.utils.Invariants.checkArgument; +import static accord.utils.Invariants.requireArgument; import static accord.utils.SortedArrays.Search.CEIL; import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer; import static org.apache.cassandra.service.accord.AccordSerializers.serialize; @@ -89,7 +89,7 @@ public class TxnUpdate extends AccordUpdate public TxnUpdate(List fragments, TxnCondition condition, @Nullable ConsistencyLevel cassandraCommitCL, boolean preserveTimestamps) { - checkArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL)); + requireArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL)); // TODO: Figure out a way to shove keys into TxnCondition, and have it implement slice/merge. this.keys = Keys.of(fragments, fragment -> fragment.key); fragments.sort(TxnWrite.Fragment::compareKeys); diff --git a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java index 7c5747003a..475c1d0853 100644 --- a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java +++ b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java @@ -188,7 +188,7 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor break; if (current.epoch.isEqualOrBefore(start)) { - Invariants.checkState(current.epoch.isDirectlyAfter(metadata.epoch)); + Invariants.require(current.epoch.isDirectlyAfter(metadata.epoch)); metadata = current.transform.execute(metadata).success().metadata; } else if (current.epoch.isAfter(start)) diff --git a/src/java/org/apache/cassandra/tcm/Processor.java b/src/java/org/apache/cassandra/tcm/Processor.java index b370a7c27e..2d8795ed91 100644 --- a/src/java/org/apache/cassandra/tcm/Processor.java +++ b/src/java/org/apache/cassandra/tcm/Processor.java @@ -132,7 +132,7 @@ public interface Processor cms.add(acc); for (Entry entry : logState.entries) { - Invariants.checkState(entry.epoch.isDirectlyAfter(acc.epoch), "%s should have been directly after %s", entry.epoch, acc.epoch); + Invariants.require(entry.epoch.isDirectlyAfter(acc.epoch), "%s should have been directly after %s", entry.epoch, acc.epoch); Transformation.Result res = entry.transform.execute(acc); assert res.isSuccess() : res.toString(); acc = res.success().metadata; diff --git a/src/java/org/apache/cassandra/tcm/log/LogReader.java b/src/java/org/apache/cassandra/tcm/log/LogReader.java index 92b57a0575..4ec597a311 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogReader.java +++ b/src/java/org/apache/cassandra/tcm/log/LogReader.java @@ -161,7 +161,7 @@ public interface LogReader } else { - Invariants.checkState(closestSnapshot.epoch.isEqualOrAfter(start), + Invariants.require(closestSnapshot.epoch.isEqualOrAfter(start), "Got %s, but requested snapshot of %s", closestSnapshot.epoch, start); EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end); return new LogState(closestSnapshot, ImmutableList.copyOf(entryHolder.entries)); diff --git a/src/java/org/apache/cassandra/tcm/log/LogState.java b/src/java/org/apache/cassandra/tcm/log/LogState.java index fcb1a05a51..96514a740c 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogState.java +++ b/src/java/org/apache/cassandra/tcm/log/LogState.java @@ -73,7 +73,7 @@ public class LogState // Uses Replication rather than an just a list of entries primarily to avoid duplicating the existing serializer public LogState(ClusterMetadata baseState, ImmutableList entries) { - Invariants.checkState(baseState == null || + Invariants.require(baseState == null || entries.isEmpty() || entries.get(0).epoch.isDirectlyAfter(baseState.epoch), "Base state: %s, first entry: %s", baseState == null ? null : baseState.epoch, entries.isEmpty() ? null : entries.get(0).epoch); diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java index 7dde51fd9f..0f6cf6eb14 100644 --- a/src/java/org/apache/cassandra/tcm/membership/Directory.java +++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java @@ -300,7 +300,7 @@ public class Directory implements MetadataValue public Directory removed(Epoch removedIn, NodeId id, InetAddressAndPort addr) { - Invariants.checkState(!peers.containsKey(id)); + Invariants.require(!peers.containsKey(id)); return new Directory(nextId, lastModified, peers, diff --git a/src/java/org/apache/cassandra/utils/MergeIterator.java b/src/java/org/apache/cassandra/utils/MergeIterator.java index 9818e93083..fdbae53d7f 100644 --- a/src/java/org/apache/cassandra/utils/MergeIterator.java +++ b/src/java/org/apache/cassandra/utils/MergeIterator.java @@ -67,7 +67,7 @@ public abstract class MergeIterator extends AbstractIterator implem @Override protected E getReduced() { - Invariants.checkState(first != null); + Invariants.require(first != null); return first; } }); diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java index 8ccecb1066..ec66bd82d5 100644 --- a/src/java/org/apache/cassandra/utils/btree/BTree.java +++ b/src/java/org/apache/cassandra/utils/btree/BTree.java @@ -143,8 +143,8 @@ public class BTree public static Object[] unsafeAllocateNonEmptyLeaf(int size) { - Invariants.checkArgument(size > 0, "size should be non-zero"); - Invariants.checkArgument(size <= MAX_KEYS, "size (%s) should be no more than %s", size, MAX_KEYS); + Invariants.requireArgument(size > 0, "size should be non-zero"); + Invariants.requireArgument(size <= MAX_KEYS, "size (%s) should be no more than %s", size, MAX_KEYS); return new Object[size | 1]; } diff --git a/src/java/org/apache/cassandra/utils/concurrent/LockWithAsyncSignal.java b/src/java/org/apache/cassandra/utils/concurrent/LockWithAsyncSignal.java index e64a87221f..87a06ea6ed 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/LockWithAsyncSignal.java +++ b/src/java/org/apache/cassandra/utils/concurrent/LockWithAsyncSignal.java @@ -107,7 +107,7 @@ public class LockWithAsyncSignal implements Lock { Thread thread = Thread.currentThread(); int restoreDepth = depth; - Invariants.checkState(owner == thread); + Invariants.require(owner == thread); depth = 0; owner = null; @@ -117,7 +117,7 @@ public class LockWithAsyncSignal implements Lock public void unlock() { - Invariants.checkState(owner == Thread.currentThread()); + Invariants.require(owner == Thread.currentThread()); if (--depth > 0) return; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index a15dc9c59a..d623bef85f 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -158,7 +158,7 @@ public class Coordinator implements ICoordinator boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue)); prepared.validate(clientState); - Invariants.checkState(prepared instanceof SelectStatement, "Only SELECT statements can be executed with paging %s", prepared); + Invariants.require(prepared instanceof SelectStatement, "Only SELECT statements can be executed with paging %s", prepared); Dispatcher.RequestTime requestTime = Dispatcher.RequestTime.forImmediateExecution(); SelectStatement selectStatement = (SelectStatement) prepared; diff --git a/test/distributed/org/apache/cassandra/distributed/test/IntegrationTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/IntegrationTestBase.java index 16f4e58f30..e05b52e06d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/IntegrationTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/IntegrationTestBase.java @@ -44,7 +44,7 @@ public class IntegrationTestBase extends TestBaseImpl protected static void init(int nodes, Consumer cfg) throws Throwable { - Invariants.checkState(!initialized); + Invariants.require(!initialized); cluster = Cluster.build() .withNodes(nodes) .withConfig(cfg) diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java index 9bc25afd85..344d529baa 100644 --- a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java +++ b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java @@ -530,35 +530,35 @@ public abstract class TopologyMixupTestBase { - c.with(Feature.values()) - .set("write_request_timeout", "10s") - .set("read_request_timeout", "10s") - .set("range_request_timeout", "20s") - .set("request_timeout", "20s") - .set("transaction_timeout", "15s") - .set("native_transport_timeout", "30s") - // bound startup to some value larger than the task timeout, this is to allow the - // tests to stop blocking when a startup issue is detected. The main reason for - // this is that startup blocks forever, waiting for accord and streaming to - // complete... but if there are bugs at these layers then the startup will never - // exit, blocking the JVM from giving the needed information (logs/seed) to debug. - .set(Constants.KEY_DTEST_STARTUP_TIMEOUT, "4m") - .set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false); - //TODO (maintenance): where to put this? Anything touching ConfigGenBuilder with jvm-dtest needs this... - ((InstanceConfig) c).remove("commitlog_sync_period_in_ms"); - for (Map.Entry e : yamlConfigOverrides.entrySet()) - c.set(e.getKey(), e.getValue()); - onConfigure(c); - }) - //TODO (maintenance): should TopologyHistory also be a INodeProvisionStrategy.Factory so address information is stored in the Node? - //TODO (maintenance): AbstractCluster's Map nodeIdTopology makes playing with dc/rack annoying, if this becomes an interface then TopologyHistory could own - .withNodeProvisionStrategy((subnet, portMap) -> new INodeProvisionStrategy.AbstractNodeProvisionStrategy(portMap) - { - { - Invariants.checkArgument(subnet == 0, "Unexpected subnet detected: %d", subnet); - } + .withTokenSupplier(topologyHistory) + .withConfig(c -> { + c.with(Feature.values()) + .set("write_request_timeout", "10s") + .set("read_request_timeout", "10s") + .set("range_request_timeout", "20s") + .set("request_timeout", "20s") + .set("transaction_timeout", "15s") + .set("native_transport_timeout", "30s") + // bound startup to some value larger than the task timeout, this is to allow the + // tests to stop blocking when a startup issue is detected. The main reason for + // this is that startup blocks forever, waiting for accord and streaming to + // complete... but if there are bugs at these layers then the startup will never + // exit, blocking the JVM from giving the needed information (logs/seed) to debug. + .set(Constants.KEY_DTEST_STARTUP_TIMEOUT, "4m") + .set(Constants.KEY_DTEST_API_STARTUP_FAILURE_AS_SHUTDOWN, false); + //TODO (maintenance): where to put this? Anything touching ConfigGenBuilder with jvm-dtest needs this... + ((InstanceConfig) c).remove("commitlog_sync_period_in_ms"); + for (Map.Entry e : yamlConfigOverrides.entrySet()) + c.set(e.getKey(), e.getValue()); + onConfigure(c); + }) + //TODO (maintenance): should TopologyHistory also be a INodeProvisionStrategy.Factory so address information is stored in the Node? + //TODO (maintenance): AbstractCluster's Map nodeIdTopology makes playing with dc/rack annoying, if this becomes an interface then TopologyHistory could own + .withNodeProvisionStrategy((subnet, portMap) -> new INodeProvisionStrategy.AbstractNodeProvisionStrategy(portMap) + { + { + Invariants.requireArgument(subnet == 0, "Unexpected subnet detected: %d", subnet); + } private final String ipPrefix = "127.0." + subnet + '.'; @@ -982,7 +982,7 @@ public abstract class TopologyMixupTestBase actual = new ArrayList<>(); // TODO: Have never tested with multiple - Invariants.checkState(visit.operations.length == 1); + Invariants.require(visit.operations.length == 1); for (UntypedResultSet.Row row : execute.apply(statement)) actual.add(resultSetToRow(schema, (Operations.SelectStatement) visit.operations[0], row)); return actual; @@ -99,7 +99,7 @@ public class CQLTesterVisitExecutor extends CQLVisitExecutor { for (int j = 0; j < schema.clusteringKeys.size(); j++) { - Invariants.checkState(!row.has(schema.clusteringKeys.get(j).name), + Invariants.require(!row.has(schema.clusteringKeys.get(j).name), "All elements of clustering key should have been null"); } clusteringKey = NIL_KEY; diff --git a/test/harry/main/org/apache/cassandra/harry/execution/CQLVisitExecutor.java b/test/harry/main/org/apache/cassandra/harry/execution/CQLVisitExecutor.java index 92a9627683..dc4eb227c4 100644 --- a/test/harry/main/org/apache/cassandra/harry/execution/CQLVisitExecutor.java +++ b/test/harry/main/org/apache/cassandra/harry/execution/CQLVisitExecutor.java @@ -139,7 +139,7 @@ public abstract class CQLVisitExecutor // All operations are not touching any data if (compiledStatement == null) { - Invariants.checkArgument(Arrays.stream(visit.operations).allMatch(op -> op.kind() == Operations.Kind.CUSTOM)); + Invariants.requireArgument(Arrays.stream(visit.operations).allMatch(op -> op.kind() == Operations.Kind.CUSTOM)); return; } @@ -150,7 +150,7 @@ public abstract class CQLVisitExecutor } else { - Invariants.checkState(selects.size() == 1); + Invariants.require(selects.size() == 1); executeValidatingVisit(visit, selects, compiledStatement); } dataTracker.end(visit); diff --git a/test/harry/main/org/apache/cassandra/harry/execution/DataTracker.java b/test/harry/main/org/apache/cassandra/harry/execution/DataTracker.java index 37b40f5acc..c27d52385e 100644 --- a/test/harry/main/org/apache/cassandra/harry/execution/DataTracker.java +++ b/test/harry/main/org/apache/cassandra/harry/execution/DataTracker.java @@ -69,7 +69,7 @@ public interface DataTracker public void begin(Visit visit) { long prev = started.get(); - Invariants.checkState(prev == 0 || visit.lts == (prev + 1)); + Invariants.require(prev == 0 || visit.lts == (prev + 1)); started.set(visit.lts); for (int i = 0; i < visit.operations.length; i++) { @@ -88,7 +88,7 @@ public interface DataTracker public void end(Visit visit) { long current = started.get(); - Invariants.checkState(current == visit.lts, "Current stated %d, current visit: %d", current, visit.lts); + Invariants.require(current == visit.lts, "Current stated %d, current visit: %d", current, visit.lts); finished.set(visit.lts); } diff --git a/test/harry/main/org/apache/cassandra/harry/execution/InJvmDTestVisitExecutor.java b/test/harry/main/org/apache/cassandra/harry/execution/InJvmDTestVisitExecutor.java index 4c4d420d8a..282e02505c 100644 --- a/test/harry/main/org/apache/cassandra/harry/execution/InJvmDTestVisitExecutor.java +++ b/test/harry/main/org/apache/cassandra/harry/execution/InJvmDTestVisitExecutor.java @@ -103,7 +103,7 @@ public class InJvmDTestVisitExecutor extends CQLVisitExecutor protected List executeWithResult(Visit visit, int node, int pageSize, CompiledStatement statement, ConsistencyLevel consistencyLevel) { - Invariants.checkState(visit.operations.length == 1); + Invariants.require(visit.operations.length == 1); Object[][] rows; if (consistencyLevel == ConsistencyLevel.NODE_LOCAL) rows = cluster.get(node).executeInternal(statement.cql(), statement.bindings()); @@ -199,7 +199,7 @@ public class InJvmDTestVisitExecutor extends CQLVisitExecutor { for (int j = 0; j < schema.clusteringKeys.size(); j++) { - Invariants.checkState(result[selection.indexOf(schema.clusteringKeys.get(j))] == null, + Invariants.require(result[selection.indexOf(schema.clusteringKeys.get(j))] == null, "All elements of clustering key should have been null"); } clusteringKey = NIL_KEY; diff --git a/test/harry/main/org/apache/cassandra/harry/execution/QueryBuildingVisitExecutor.java b/test/harry/main/org/apache/cassandra/harry/execution/QueryBuildingVisitExecutor.java index a330007a2a..e791f6f12c 100644 --- a/test/harry/main/org/apache/cassandra/harry/execution/QueryBuildingVisitExecutor.java +++ b/test/harry/main/org/apache/cassandra/harry/execution/QueryBuildingVisitExecutor.java @@ -72,7 +72,7 @@ public class QueryBuildingVisitExecutor extends VisitExecutor return query; } - Invariants.checkState(bindings.isEmpty() && visitedPds.isEmpty() && selects.isEmpty()); + Invariants.require(bindings.isEmpty() && visitedPds.isEmpty() && selects.isEmpty()); return null; } @@ -107,7 +107,7 @@ public class QueryBuildingVisitExecutor extends VisitExecutor { if (statements.isEmpty()) { - Invariants.checkState(bindings.isEmpty() && visitedPds.isEmpty() && selects.isEmpty()); + Invariants.require(bindings.isEmpty() && visitedPds.isEmpty() && selects.isEmpty()); return; } diff --git a/test/harry/main/org/apache/cassandra/harry/execution/RingAwareInJvmDTestVisitExecutor.java b/test/harry/main/org/apache/cassandra/harry/execution/RingAwareInJvmDTestVisitExecutor.java index b9a6e0565d..5489707bad 100644 --- a/test/harry/main/org/apache/cassandra/harry/execution/RingAwareInJvmDTestVisitExecutor.java +++ b/test/harry/main/org/apache/cassandra/harry/execution/RingAwareInJvmDTestVisitExecutor.java @@ -119,7 +119,7 @@ public class RingAwareInJvmDTestVisitExecutor extends InJvmDTestVisitExecutor { try { - Invariants.checkState(visit.visitedPartitions.size() == 1, + Invariants.require(visit.visitedPartitions.size() == 1, "Ring aware executor can only read and write one partition at a time"); for (TokenPlacementModel.Replica replica : getReplicasFor(visit.visitedPartitions.iterator().next().longValue())) { diff --git a/test/harry/main/org/apache/cassandra/harry/gen/Generators.java b/test/harry/main/org/apache/cassandra/harry/gen/Generators.java index f1527fc8a9..9bffc76475 100644 --- a/test/harry/main/org/apache/cassandra/harry/gen/Generators.java +++ b/test/harry/main/org/apache/cassandra/harry/gen/Generators.java @@ -282,7 +282,7 @@ public class Generators { T v = delegate.generate(rng); int hashCode = v.hashCode(); - Invariants.checkState(hashCode != System.identityHashCode(v), "hashCode was not overridden for type %s", v.getClass()); + Invariants.require(hashCode != System.identityHashCode(v), "hashCode was not overridden for type %s", v.getClass()); if (hashCodes.contains(hashCode)) continue; hashCodes.add(hashCode); diff --git a/test/harry/main/org/apache/cassandra/harry/gen/InvertibleGenerator.java b/test/harry/main/org/apache/cassandra/harry/gen/InvertibleGenerator.java index ae8f4a8f71..f16185da37 100644 --- a/test/harry/main/org/apache/cassandra/harry/gen/InvertibleGenerator.java +++ b/test/harry/main/org/apache/cassandra/harry/gen/InvertibleGenerator.java @@ -74,9 +74,9 @@ public class InvertibleGenerator implements HistoryBuilder.IndexedBijection gen, Comparator comparator) { - Invariants.checkState(population > 0, + Invariants.require(population > 0, "Population should be strictly positive %d", population); - Invariants.checkState(Long.compareUnsigned(typeEntropy, 0) > 0, + Invariants.require(Long.compareUnsigned(typeEntropy, 0) > 0, "Type entropy should be strictly positive, but was %d: %s", typeEntropy, gen); // We can / will generate at most that many values @@ -101,7 +101,7 @@ public class InvertibleGenerator implements HistoryBuilder.IndexedBijection implements HistoryBuilder.IndexedBijection 0, + Invariants.require( comparator.compare(current, prev) > 0, () -> String.format("%s should be strictly after %s", prev, current)); } } @@ -138,7 +138,7 @@ public class InvertibleGenerator implements HistoryBuilder.IndexedBijection implements HistoryBuilder.IndexedBijection implements HistoryBuilder.IndexedBijection, Integer> columns = new HashMap<>(); for (int i = 0; i < schema.allColumnInSelectOrder.size(); i++) { diff --git a/test/harry/main/org/apache/cassandra/harry/test/SimpleBijectionTest.java b/test/harry/main/org/apache/cassandra/harry/test/SimpleBijectionTest.java index fde20aa363..9e090895da 100644 --- a/test/harry/main/org/apache/cassandra/harry/test/SimpleBijectionTest.java +++ b/test/harry/main/org/apache/cassandra/harry/test/SimpleBijectionTest.java @@ -52,7 +52,7 @@ public class SimpleBijectionTest Object next = generator.inflate(generator.descriptorAt(i)); if (previous != null) { - Invariants.checkState(column.type.comparator().compare(next, previous) > 0, + Invariants.require(column.type.comparator().compare(next, previous) > 0, "%s should be > %s", next, previous); } previous = next; diff --git a/test/simulator/main/org/apache/cassandra/simulator/logging/RunStartDefiner.java b/test/simulator/main/org/apache/cassandra/simulator/logging/RunStartDefiner.java index 92066c1824..6d541eb7c3 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/logging/RunStartDefiner.java +++ b/test/simulator/main/org/apache/cassandra/simulator/logging/RunStartDefiner.java @@ -26,7 +26,7 @@ public class RunStartDefiner extends PropertyDefinerBase { static { - Invariants.checkState(CassandraRelevantProperties.SIMULATOR_STARTED.getString() != null); + Invariants.require(CassandraRelevantProperties.SIMULATOR_STARTED.getString() != null); } @Override diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/HarrySimulatorTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/HarrySimulatorTest.java index 61fb38fc7e..14073943d7 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/HarrySimulatorTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/HarrySimulatorTest.java @@ -757,7 +757,7 @@ public class HarrySimulatorTest Visit visit = new Visit(lts, new Operations.Operation[]{ simulation.insertGen.generate(simulation.rng).toOp(lts) }); Visit prev_ = simulation.log.put(lts, visit); - Invariants.checkState(prev_ == null); + Invariants.require(prev_ == null); actions[i] = new Actions.LambdaAction("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, () -> { CompiledStatement compiledStatement = simulation.queryBuilder.compile(visit); diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/HarryValidatingQuery.java b/test/simulator/test/org/apache/cassandra/simulator/test/HarryValidatingQuery.java index 2a011c944c..72cd85de3b 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/HarryValidatingQuery.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/HarryValidatingQuery.java @@ -79,8 +79,8 @@ public class HarryValidatingQuery extends SimulatedAction try { TokenPlacementModel.ReplicatedRanges ring = rf.replicate(owernship); - Invariants.checkState(visit.operations.length == 1); - Invariants.checkState(visit.operations[0] instanceof Operations.SelectStatement); + Invariants.require(visit.operations.length == 1); + Invariants.require(visit.operations[0] instanceof Operations.SelectStatement); Operations.SelectStatement select = (Operations.SelectStatement) visit.operations[0]; for (TokenPlacementModel.Replica replica : ring.replicasFor(token(select.pd))) { diff --git a/test/unit/org/apache/cassandra/concurrent/ManualExecutor.java b/test/unit/org/apache/cassandra/concurrent/ManualExecutor.java index 5600e3f3c1..2b98d13d6a 100644 --- a/test/unit/org/apache/cassandra/concurrent/ManualExecutor.java +++ b/test/unit/org/apache/cassandra/concurrent/ManualExecutor.java @@ -110,7 +110,7 @@ public class ManualExecutor implements ExecutorPlus Task(Runnable runnable, Callable callable, WithResources withResources, Object result, FutureImpl future) { - Invariants.checkArgument(runnable != null ^ callable != null); + Invariants.requireArgument(runnable != null ^ callable != null); this.runnable = runnable; this.callable = callable; diff --git a/test/unit/org/apache/cassandra/cql3/ast/Visitor.java b/test/unit/org/apache/cassandra/cql3/ast/Visitor.java index c87415e981..36fb3f33d6 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/Visitor.java +++ b/test/unit/org/apache/cassandra/cql3/ast/Visitor.java @@ -92,7 +92,7 @@ public interface Visitor public static CompositeVisitor of(List visitors) { - Invariants.checkArgument(!visitors.isEmpty(), "Visitors may not be empty"); + Invariants.requireArgument(!visitors.isEmpty(), "Visitors may not be empty"); if (Stream.of(visitors).noneMatch(v -> v instanceof CompositeVisitor)) return new CompositeVisitor(visitors); diff --git a/test/unit/org/apache/cassandra/journal/IndexTest.java b/test/unit/org/apache/cassandra/journal/IndexTest.java index 9b6a69505e..43e2e21fa9 100644 --- a/test/unit/org/apache/cassandra/journal/IndexTest.java +++ b/test/unit/org/apache/cassandra/journal/IndexTest.java @@ -218,7 +218,7 @@ public class IndexTest for (long i : value) inMemory.update(key, Index.readOffset(i), Index.readSize(i)); for (int i = 1 ; i < value.length ; ++i) - Invariants.checkState(value[i - 1] > value[i]); + Invariants.require(value[i - 1] > value[i]); } assertIndex(map, inMemory); diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java index 14f5b67056..0efac1d081 100644 --- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java +++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java @@ -336,7 +336,7 @@ public class EpochSyncTest if (cms.metadata().inProgressSequences.isEmpty()) throw new IllegalStateException("Attempted to bump epoch when nothing was pending"); Iterator> it = cms.metadata().inProgressSequences.iterator(); - Invariants.checkState(it.hasNext()); + Invariants.require(it.hasNext()); notify(process(it.next()).metadata); } @@ -562,8 +562,8 @@ public class EpochSyncTest void registerNode(Node.Id id, long token) { - Invariants.checkState(!tokens.contains(token), "Attempted to add token %d for node %s but token is already taken", token, id); - Invariants.checkState(!instances.containsKey(id), "Attempted to add node %s; but already exists", id); + Invariants.require(!tokens.contains(token), "Attempted to add token %d for node %s but token is already taken", token, id); + Invariants.require(!instances.containsKey(id), "Attempted to add node %s; but already exists", id); ClusterMetadata.Transformer builder = cms.metadata().transformer(); @@ -598,7 +598,7 @@ public class EpochSyncTest void removeNode(Node.Id pick) { Instance inst = Objects.requireNonNull(instances.get(pick), "Unknown id " + pick); - Invariants.checkState(!removed.contains(pick), "Can not remove node twice; node " + pick); + Invariants.require(!removed.contains(pick), "Can not remove node twice; node " + pick); removed.add(pick); inst.status = Status.Leaving; PrepareLeave prepareLeave = new PrepareLeave(new NodeId(pick.id), false, new UniformRangePlacement(), LeaveStreams.Kind.REMOVENODE); @@ -747,19 +747,19 @@ public class EpochSyncTest switch (status) { case Init: - Invariants.checkState(!t.nodes().contains(id), "Node was in Init state but present in the Topology!"); - Invariants.checkState(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); + Invariants.require(!t.nodes().contains(id), "Node was in Init state but present in the Topology!"); + Invariants.require(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); start(); status = Status.Registered; break; case Registered: - Invariants.checkState(!t.nodes().contains(id), "Node was in Init state but present in the Topology!"); - Invariants.checkState(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); + Invariants.require(!t.nodes().contains(id), "Node was in Init state but present in the Topology!"); + Invariants.require(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); if (current.placements.get(replication_params).writes.byEndpoint().keySet().contains(address(id))) status = Status.Joining; break; case Joining: - Invariants.checkState(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); + Invariants.require(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); if (joined(current, id)) status = Status.Joined; case Removed: diff --git a/test/unit/org/apache/cassandra/service/accord/MockDiskStateManager.java b/test/unit/org/apache/cassandra/service/accord/MockDiskStateManager.java index 88178a55ab..c3b9bbf433 100644 --- a/test/unit/org/apache/cassandra/service/accord/MockDiskStateManager.java +++ b/test/unit/org/apache/cassandra/service/accord/MockDiskStateManager.java @@ -87,7 +87,7 @@ public enum MockDiskStateManager implements AccordConfigurationService.DiskState { if (diskState.isEmpty()) return AccordKeyspace.EpochDiskState.create(epoch); - Invariants.checkArgument(epoch >= diskState.minEpoch, "Epoch %d < %d (min)", epoch, diskState.minEpoch); + Invariants.requireArgument(epoch >= diskState.minEpoch, "Epoch %d < %d (min)", epoch, diskState.minEpoch); if (epoch > diskState.maxEpoch) diskState = diskState.withNewMaxEpoch(epoch); return diskState; diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index 4b35621570..ad8a995794 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -351,7 +351,7 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester protected static Gen>> randomTxn(Gen domainGen, Gen.LongGen tokenGen) { TableMetadata tbl = reverseTokenTbl; - Invariants.checkArgument(tbl.partitioner == Murmur3Partitioner.instance, "Only murmur partitioner is supported; given %s", tbl.partitioner.getClass()); + Invariants.requireArgument(tbl.partitioner == Murmur3Partitioner.instance, "Only murmur partitioner is supported; given %s", tbl.partitioner.getClass()); Gen keyGen = rs -> new PartitionKey(tbl.id, tbl.partitioner.decorateKey(Murmur3Partitioner.LongToken.keyForToken(tokenGen.nextLong(rs)))); Gen rangeGen = rs -> { long a = tokenGen.nextLong(rs); diff --git a/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java b/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java index ac31fa97a9..aabbfbfb6d 100644 --- a/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java +++ b/test/unit/org/apache/cassandra/utils/SimulatedMiniCluster.java @@ -253,7 +253,7 @@ public class SimulatedMiniCluster if (current.inProgressSequences.isEmpty()) throw new IllegalStateException("Attempted to bump epoch when nothing was pending"); Iterator> it = current.inProgressSequences.iterator(); - Invariants.checkState(it.hasNext()); + Invariants.require(it.hasNext()); notifyMetadataChange(process(it.next()).metadata); }