diff --git a/modules/accord b/modules/accord index c076383eb4..c80ec466f5 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c076383eb432670c4d919e9fd0db76296169ea00 +Subproject commit c80ec466f5dd633d4fcbaedc5674f183446def34 diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index 0b8bc8dd71..ec8a37b956 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -96,7 +96,6 @@ import org.apache.cassandra.service.accord.serializers.EnumSerializer; import org.apache.cassandra.service.accord.serializers.FetchSerializers; import org.apache.cassandra.service.accord.serializers.CalculateDepsSerializers; import org.apache.cassandra.service.accord.serializers.GetEphmrlReadDepsSerializers; -import org.apache.cassandra.service.accord.serializers.GetMaxConflictSerializers; import org.apache.cassandra.service.accord.serializers.InformDurableSerializers; import org.apache.cassandra.service.accord.serializers.LatestDepsSerializers; import org.apache.cassandra.service.accord.serializers.PreacceptSerializers; @@ -341,8 +340,6 @@ public enum Verb ACCORD_GET_LATEST_DEPS_REQ (168, P2, writeTimeout, IMMEDIATE, () -> LatestDepsSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_GET_LATEST_DEPS_RSP), ACCORD_GET_EPHMRL_READ_DEPS_RSP (161, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.reply, AccordService::responseHandlerOrNoop ), ACCORD_GET_EPHMRL_READ_DEPS_REQ (162, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_GET_EPHMRL_READ_DEPS_RSP), - ACCORD_GET_MAX_CONFLICT_RSP (163, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.reply, AccordService::responseHandlerOrNoop ), - ACCORD_GET_MAX_CONFLICT_REQ (164, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP), ACCORD_FETCH_DATA_RSP (145, P2, writeTimeout, IMMEDIATE, () -> FetchSerializers.reply, AccordService::responseHandlerOrNoop ), ACCORD_FETCH_DATA_REQ (146, P2, writeTimeout, IMMEDIATE, () -> FetchSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_FETCH_DATA_RSP ), ACCORD_SET_SHARD_DURABLE_REQ (147, P2, writeTimeout, MISC, () -> SetDurableSerializers.shardDurable, AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ), diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 6e45b6bbb3..667199c6e1 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -181,10 +181,10 @@ public class AccordCommandStore extends CommandStore this.commandsForRanges = new CommandsForRanges.Manager(this); this.loader = new CommandStoreLoader(this); - loadRedundantBefore(journal.loadRedundantBefore(id())); - loadBootstrapBeganAt(journal.loadBootstrapBeganAt(id())); - loadSafeToRead(journal.loadSafeToRead(id())); - loadRangesForEpoch(journal.loadRangesForEpoch(id())); + maybeLoadRedundantBefore(journal.loadRedundantBefore(id())); + maybeLoadBootstrapBeganAt(journal.loadBootstrapBeganAt(id())); + maybeLoadSafeToRead(journal.loadSafeToRead(id())); + maybeLoadRangesForEpoch(journal.loadRangesForEpoch(id())); } static Factory factory(Journal journal, IntFunction executorFactory) @@ -255,13 +255,6 @@ public class AccordCommandStore extends CommandStore return caches; } - @VisibleForTesting - @Override - public void unsafeSetRangesForEpoch(CommandStores.RangesForEpoch newRangesForEpoch) - { - super.unsafeSetRangesForEpoch(newRangesForEpoch); - } - @Nullable @VisibleForTesting public Runnable appendToKeyspace(TxnId txnId, Command after) @@ -556,27 +549,27 @@ public class AccordCommandStore extends CommandStore * Replay/state reloading */ - void loadRedundantBefore(RedundantBefore redundantBefore) + void maybeLoadRedundantBefore(RedundantBefore redundantBefore) { if (redundantBefore != null) - unsafeSetRedundantBefore(redundantBefore); + loadRedundantBefore(redundantBefore); } - void loadBootstrapBeganAt(NavigableMap bootstrapBeganAt) + void maybeLoadBootstrapBeganAt(NavigableMap bootstrapBeganAt) { if (bootstrapBeganAt != null) - unsafeSetBootstrapBeganAt(bootstrapBeganAt); + loadBootstrapBeganAt(bootstrapBeganAt); } - void loadSafeToRead(NavigableMap safeToRead) + void maybeLoadSafeToRead(NavigableMap safeToRead) { if (safeToRead != null) - unsafeSetSafeToRead(safeToRead); + loadSafeToRead(safeToRead); } - void loadRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch) + void maybeLoadRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch) { if (rangesForEpoch != null) - unsafeSetRangesForEpoch(rangesForEpoch); + loadRangesForEpoch(rangesForEpoch); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java index 5f41107f18..9d91e386e8 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java @@ -34,6 +34,7 @@ import accord.impl.AbstractFetchCoordinator; import accord.local.CommandStore; import accord.local.Node; import accord.local.SafeCommandStore; +import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.Participants; import accord.primitives.Range; @@ -44,6 +45,7 @@ import accord.primitives.Seekables; import accord.primitives.SyncPoint; import accord.primitives.Timestamp; import accord.primitives.Txn; +import accord.primitives.TxnId; import accord.utils.Invariants; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; @@ -267,13 +269,13 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements Invariants.checkArgument(key.domain() == Routable.Domain.Range, "Required Range but saw %s: %s", key.domain(), key); TokenRange range = (TokenRange) key; - // TODO (correctness): check epoch - // TODO (correctness): handle dropped tables + // 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); - // FIXME: may also be relocation + // TODO (required): may also be relocation StreamPlan plan = new StreamPlan(StreamOperation.BOOTSTRAP, 1, false, null, PreviewKind.NONE).flushBeforeTransfer(true); @@ -410,4 +412,27 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements } }); } + + public static class AccordFetchRequest extends FetchRequest + { + public AccordFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn) + { + super(sourceEpoch, syncId, ranges, partialDeps, partialTxn); + } + + @Override + protected AsyncChain beginRead(SafeCommandStore safeStore, Timestamp executeAt, PartialTxn txn, Ranges unavailable) + { + AsyncChain result = super.beginRead(safeStore, executeAt, txn, unavailable); + // TODO (required): verify that streaming snapshots have all been created by now, so we won't stream any data that arrives after this + readStarted(safeStore, unavailable); + return result; + } + } + + @Override + protected FetchRequest newFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn) + { + return new AccordFetchRequest(sourceEpoch, syncId, ranges, partialDeps, partialTxn); + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index f82b0b3e4d..078e596ce2 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -33,6 +33,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.impl.CommandChange; +import accord.impl.CommandChange.Field; import accord.impl.ErasedSafeCommand; import accord.local.Cleanup; import accord.local.Command; @@ -543,7 +544,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable int iterable = toIterableSetFields(flags); while (iterable != 0) { - CommandChange.Fields field = nextSetField(iterable); + Field field = nextSetField(iterable); if (getFieldIsNull(field, flags)) { iterable = unsetIterableFields(field, iterable); @@ -653,7 +654,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable int iterable = toIterableSetFields(flags); while (iterable != 0) { - CommandChange.Fields field = nextSetField(iterable); + Field field = nextSetField(iterable); if (getFieldChanged(field, this.flags) || getFieldIsNull(field, mask)) { if (!getFieldIsNull(field, flags)) @@ -677,7 +678,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable } } - private void deserialize(CommandChange.Fields field, DataInputPlus in, int userVersion) throws IOException + private void deserialize(Field field, DataInputPlus in, int userVersion) throws IOException { switch (field) { @@ -740,7 +741,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable } } - private void skip(CommandChange.Fields field, DataInputPlus in, int userVersion) throws IOException + private void skip(Field field, DataInputPlus in, int userVersion) throws IOException { switch (field) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java index cff8ca2d96..afa2cfcc6c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java +++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java @@ -141,8 +141,6 @@ public class AccordMessageSink implements MessageSink builder.put(MessageType.GET_LATEST_DEPS_RSP, Verb.ACCORD_GET_LATEST_DEPS_RSP); builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_REQ, Verb.ACCORD_GET_EPHMRL_READ_DEPS_REQ); builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_RSP, Verb.ACCORD_GET_EPHMRL_READ_DEPS_RSP); - builder.put(MessageType.GET_MAX_CONFLICT_REQ, Verb.ACCORD_GET_MAX_CONFLICT_REQ); - builder.put(MessageType.GET_MAX_CONFLICT_RSP, Verb.ACCORD_GET_MAX_CONFLICT_RSP); builder.put(MessageType.COMMIT_SLOW_PATH_REQ, Verb.ACCORD_COMMIT_REQ); builder.put(MessageType.COMMIT_MAXIMAL_REQ, Verb.ACCORD_COMMIT_REQ); builder.put(MessageType.STABLE_FAST_PATH_REQ, Verb.ACCORD_COMMIT_REQ); diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index 1679f3da58..da811ce83a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -297,7 +297,7 @@ public class AccordObjectSizes final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true, false), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null)); final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.empty(Domain.Key), EMPTY_WRITES, ResultSerializers.APPLIED)); final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null)); - final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID)); + final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID))); private static long emptySize(Command command) { diff --git a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java index 2bff57054e..34a0df8f6b 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java @@ -30,6 +30,7 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.service.accord.AccordFetchCoordinator.AccordFetchRequest; import org.apache.cassandra.service.accord.AccordFetchCoordinator.StreamData; import org.apache.cassandra.service.accord.AccordFetchCoordinator.StreamingTxn; import org.apache.cassandra.utils.CastingSerializer; @@ -55,11 +56,11 @@ public class FetchSerializers @Override public FetchRequest deserialize(DataInputPlus in, int version) throws IOException { - return new FetchRequest(in.readUnsignedVInt(), - CommandSerializers.txnId.deserialize(in, version), - KeySerializers.ranges.deserialize(in, version), - DepsSerializers.partialDeps.deserialize(in, version), - StreamingTxn.serializer.deserialize(in, version)); + return new AccordFetchRequest(in.readUnsignedVInt(), + CommandSerializers.txnId.deserialize(in, version), + KeySerializers.ranges.deserialize(in, version), + DepsSerializers.partialDeps.deserialize(in, version), + StreamingTxn.serializer.deserialize(in, version)); } @Override @@ -91,7 +92,7 @@ public class FetchSerializers FetchResponse response = (FetchResponse) reply; serializeNullable(response.unavailable, out, version, KeySerializers.ranges); serializeNullable(response.data, out, version, streamDataSerializer); - CommandSerializers.nullableTimestamp.serialize(response.maxApplied, out, version); + CommandSerializers.nullableTimestamp.serialize(response.safeToReadAfter, out, version); } @Override @@ -116,7 +117,7 @@ public class FetchSerializers return TypeSizes.BYTE_SIZE + serializedNullableSize(response.unavailable, version, KeySerializers.ranges) + serializedNullableSize(response.data, version, streamDataSerializer) - + CommandSerializers.nullableTimestamp.serializedSize(response.maxApplied, version); + + CommandSerializers.nullableTimestamp.serializedSize(response.safeToReadAfter, version); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java deleted file mode 100644 index bb580690c4..0000000000 --- a/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java +++ /dev/null @@ -1,81 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.service.accord.serializers; - -import java.io.IOException; - -import accord.messages.GetMaxConflict; -import accord.messages.GetMaxConflict.GetMaxConflictOk; -import accord.primitives.Route; -import accord.primitives.Timestamp; -import accord.primitives.TxnId; -import org.apache.cassandra.db.TypeSizes; -import org.apache.cassandra.io.IVersionedSerializer; -import org.apache.cassandra.io.util.DataInputPlus; -import org.apache.cassandra.io.util.DataOutputPlus; - -public class GetMaxConflictSerializers -{ - public static final IVersionedSerializer request = new TxnRequestSerializer.WithUnsyncedSerializer() - { - @Override - public void serializeBody(GetMaxConflict msg, DataOutputPlus out, int version) throws IOException - { - out.writeUnsignedVInt(msg.executionEpoch); - } - - @Override - public GetMaxConflict deserializeBody(DataInputPlus in, int version, TxnId txnId, Route scope, long waitForEpoch, long minEpoch) throws IOException - { - long executionEpoch = in.readUnsignedVInt(); - return GetMaxConflict.SerializationSupport.create(scope, waitForEpoch, minEpoch, executionEpoch); - } - - @Override - public long serializedBodySize(GetMaxConflict msg, int version) - { - return TypeSizes.sizeofUnsignedVInt(msg.executionEpoch); - } - }; - - public static final IVersionedSerializer reply = new IVersionedSerializer() - { - @Override - public void serialize(GetMaxConflictOk reply, DataOutputPlus out, int version) throws IOException - { - CommandSerializers.timestamp.serialize(reply.maxConflict, out, version); - out.writeUnsignedVInt(reply.latestEpoch); - } - - @Override - public GetMaxConflictOk deserialize(DataInputPlus in, int version) throws IOException - { - Timestamp maxConflict = CommandSerializers.timestamp.deserialize(in, version); - long latestEpoch = in.readUnsignedVInt(); - return new GetMaxConflictOk(maxConflict, latestEpoch); - } - - @Override - public long serializedSize(GetMaxConflictOk reply, int version) - { - return CommandSerializers.timestamp.serializedSize(reply.maxConflict, version) - + TypeSizes.sizeofUnsignedVInt(reply.latestEpoch); - } - }; -} diff --git a/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java index 3b4acb00a5..da0d0b0077 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java @@ -333,7 +333,7 @@ public interface IVersionedWithKeysSerializer, T> extends default: throw new AssertionError("Unhandled domain: " + superset.domain()); case Key: { - RoutingKeys keys = (RoutingKeys) superset; + AbstractUnseekableKeys keys = (AbstractUnseekableKeys) superset; RoutingKey[] out = new RoutingKey[deserializeCount]; int supersetIndex = 0; int count = 0; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index 683131c1da..3c439b6e96 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -98,8 +98,8 @@ public class RecoverySerializers CommandSerializers.ballot.serialize(recoverOk.accepted, out, version); CommandSerializers.nullableTimestamp.serialize(recoverOk.executeAt, out, version); latestDeps.serialize(recoverOk.deps, out, version); - DepsSerializers.deps.serialize(recoverOk.earlierCommittedWitness, out, version); - DepsSerializers.deps.serialize(recoverOk.earlierAcceptedNoWitness, out, version); + DepsSerializers.deps.serialize(recoverOk.earlierWait, out, version); + DepsSerializers.deps.serialize(recoverOk.earlierNoWait, out, version); out.writeBoolean(recoverOk.selfAcceptsFastPath); out.writeBoolean(recoverOk.supersedingRejects); CommandSerializers.nullableWrites.serialize(recoverOk.writes, out, version); @@ -118,9 +118,9 @@ public class RecoverySerializers return new RecoverNack(kind, supersededBy); } - RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean acceptsFastPath, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) + RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierWait, Deps earlierNoWait, boolean acceptsFastPath, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) { - return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierCommittedWitness, earlierAcceptedNoWitness, acceptsFastPath, rejectsFastPath, writes, result); + return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierWait, earlierNoWait, acceptsFastPath, rejectsFastPath, writes, result); } @Override @@ -164,8 +164,8 @@ public class RecoverySerializers size += CommandSerializers.ballot.serializedSize(recoverOk.accepted, version); size += CommandSerializers.nullableTimestamp.serializedSize(recoverOk.executeAt, version); size += latestDeps.serializedSize(recoverOk.deps, version); - size += DepsSerializers.deps.serializedSize(recoverOk.earlierCommittedWitness, version); - size += DepsSerializers.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version); + size += DepsSerializers.deps.serializedSize(recoverOk.earlierWait, version); + size += DepsSerializers.deps.serializedSize(recoverOk.earlierNoWait, version); size += TypeSizes.sizeof(recoverOk.selfAcceptsFastPath); size += TypeSizes.sizeof(recoverOk.supersedingRejects); size += CommandSerializers.nullableWrites.serializedSize(recoverOk.writes, version); diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java index f07040cf6b..22ce8581dd 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java @@ -779,7 +779,7 @@ public class ClusterSimulation implements AutoCloseable .set("memtable_allocation_type", builder.memoryListener != null ? "unslabbed_heap_buffers_logged" : "heap_buffers") .set("file_cache_size", "16MiB") .set("use_deterministic_table_id", true) - .set("accord.queue_submission_model", "EXEC_ST") + .set("accord.queue_submission_model", "ASYNC") .set("disk_access_mode", "standard") .set("failure_detector", SimulatedFailureDetector.Instance.class.getName()) .set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap())) diff --git a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java index 6a9a055822..2d3551851b 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java @@ -27,6 +27,7 @@ import accord.impl.RequestCallbacks; import accord.messages.ReadData; import accord.messages.ReadData.CommitOrReadNack; import accord.topology.TopologyUtils; +import org.apache.cassandra.service.accord.AccordFetchCoordinator.AccordFetchRequest; import org.apache.cassandra.service.accord.api.AccordTimeService; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -82,7 +83,7 @@ public class AccordMessageSinkTest TxnId id = nextTxnId(epoch, txn); Ranges ranges = Ranges.of(IntKey.range(40, 50)); PartialTxn partialTxn = txn.slice(ranges, true); - Request request = new AbstractFetchCoordinator.FetchRequest(epoch, id, ranges, PartialDeps.NONE, partialTxn); + Request request = new AccordFetchRequest(epoch, id, ranges, PartialDeps.NONE, partialTxn); checkRequestReplies(request, new AbstractFetchCoordinator.FetchResponse(null, null, id), diff --git a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java index 2878c5750a..f188affea9 100644 --- a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java +++ b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java @@ -53,7 +53,7 @@ import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.p public class CommandChangeTest { - private static final EnumSet ALL = EnumSet.allOf(Fields.class); + private static final EnumSet ALL = EnumSet.allOf(Field.class); @BeforeClass public static void beforeClass() throws Throwable @@ -77,9 +77,9 @@ public class CommandChangeTest public void simpleNullChangeCheck() { int flags = getFlags(null, Command.NotDefined.uninitialised(TxnId.NONE)); - EnumSet has = EnumSet.of(Fields.SAVE_STATUS, Fields.PARTICIPANTS, Fields.DURABILITY, Fields.PROMISED, - Fields.ACCEPTED /* this is Zero... which kinda means null... */); - Set missing = Sets.difference(ALL, has); + EnumSet has = EnumSet.of(Field.SAVE_STATUS, Field.PARTICIPANTS, Field.DURABILITY, Field.PROMISED, + Field.ACCEPTED /* this is Zero... which kinda means null... */); + Set missing = Sets.difference(ALL, has); assertHas(flags, has); assertMissing(flags, missing); } @@ -117,10 +117,10 @@ public class CommandChangeTest } } - private void assertHas(int flags, Set missing) + private void assertHas(int flags, Set missing) { SoftAssertions checks = new SoftAssertions(); - for (Fields field : missing) + for (Field field : missing) { checks.assertThat(CommandChange.getFieldChanged(field, flags)) .describedAs("field %s changed", field). @@ -132,12 +132,12 @@ public class CommandChangeTest checks.assertAll(); } - private void assertMissing(int flags, Set missing) + private void assertMissing(int flags, Set missing) { SoftAssertions checks = new SoftAssertions(); - for (Fields field : missing) + for (Field field : missing) { - if (field == Fields.CLEANUP) continue; + if (field == Field.CLEANUP) continue; checks.assertThat(CommandChange.getFieldChanged(field, flags)) .describedAs("field %s changed", field) .isFalse(); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 65b315a7e2..786429bc6c 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -224,7 +224,6 @@ public class SimulatedAccordCommandStore implements AutoCloseable this.topologies = new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, topology); CommandStores.RangesForEpoch rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), topology.ranges()); updateHolder.add(topology.epoch(), rangesForEpoch, topology.ranges()); - updateHolder.updateGlobal(topology.ranges()); commandStore.unsafeUpdateRangesForEpoch(); shouldEvict = boolSource(rs.fork()); diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 4a3bc8cfd9..7d9ab80872 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -198,7 +198,7 @@ public class CommandsForKeySerializerTest return Command.SerializerSupport.executed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.empty(txnId.domain()), new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true)), new TxnData()); case Invalidated: - return Command.SerializerSupport.invalidated(txnId); + return Command.SerializerSupport.invalidated(txnId, attributes().participants()); } } diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index 4d6326b222..c551d4a023 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -273,7 +273,7 @@ public class AccordGenerators case Erased: case ErasedOrVestigial: case Invalidated: - return Command.SerializerSupport.invalidated(txnId); + return Command.SerializerSupport.invalidated(txnId, attributes(saveStatus).participants()); } } }