natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

FetchRequest should report as unavailable any slice that executes in a later epoch that is not owned by the replicas 

Improve:
 - Remove GetMaxConflict; use local MaxConflict collection

Fix:
 - Invalidated should retain StoreParticipants so we can update CFK on journal replay
 - loading pruned uninitialised commands via CFK: make sure hasTouched contains key so that if invalidated we are notified
 - updateExecuteAtLeast should always be higher than TxnId
 - Async CFK callbacks treated pruned transactions incorrectly
 - node.withEpoch when ExecuteEphemeralRead in futureEpoch
 - Deps.without should be key/range aware
 - Durably mark bootstrapBeganAt and safeToReadAt in MaxConflicts
 - Don't attempt to calculate local deps when DepsErased in GetLatestDeps (command has durably applied to this shard)
 - filter StoreParticipants before invoking shouldCleanup

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20183
This commit is contained in:
Benedict Elliott Smith 2025-01-05 21:16:25 +00:00 committed by David Capwell
parent dfcf3aff4e
commit ed2ca6e2fb
17 changed files with 76 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit c076383eb432670c4d919e9fd0db76296169ea00
Subproject commit c80ec466f5dd633d4fcbaedc5674f183446def34

3
src/java/org/apache/cassandra/net/Verb.java
View File

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

31
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -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<AccordExecutor> 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<TxnId, Ranges> bootstrapBeganAt)
void maybeLoadBootstrapBeganAt(NavigableMap<TxnId, Ranges> bootstrapBeganAt)
{
if (bootstrapBeganAt != null)
unsafeSetBootstrapBeganAt(bootstrapBeganAt);
loadBootstrapBeganAt(bootstrapBeganAt);
}
void loadSafeToRead(NavigableMap<Timestamp, Ranges> safeToRead)
void maybeLoadSafeToRead(NavigableMap<Timestamp, Ranges> 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);
}
}

31
src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
View File

@ -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<Data> beginRead(SafeCommandStore safeStore, Timestamp executeAt, PartialTxn txn, Ranges unavailable)
{
AsyncChain<Data> 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);
}
}

9
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

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

2
src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
View File

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

2
src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
View File

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

15
src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
View File

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

81
src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java
View File

@ -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<GetMaxConflict> request = new TxnRequestSerializer.WithUnsyncedSerializer<GetMaxConflict>()
{
@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<GetMaxConflictOk> reply = new IVersionedSerializer<GetMaxConflictOk>()
{
@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);
}
};
}

2
src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java
View File

@ -333,7 +333,7 @@ public interface IVersionedWithKeysSerializer<K extends Routables<?>, 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;

12
src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
View File

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

2
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
View File

@ -779,7 +779,7 @@ public class ClusterSimulation<S extends Simulation> 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()))

3
test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
View File

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

18
test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java
View File

@ -53,7 +53,7 @@ import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.p
public class CommandChangeTest
{
private static final EnumSet<Fields> ALL = EnumSet.allOf(Fields.class);
private static final EnumSet<Field> 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<Fields> has = EnumSet.of(Fields.SAVE_STATUS, Fields.PARTICIPANTS, Fields.DURABILITY, Fields.PROMISED,
Fields.ACCEPTED /* this is Zero... which kinda means null... */);
Set<Fields> missing = Sets.difference(ALL, has);
EnumSet<Field> has = EnumSet.of(Field.SAVE_STATUS, Field.PARTICIPANTS, Field.DURABILITY, Field.PROMISED,
Field.ACCEPTED /* this is Zero... which kinda means null... */);
Set<Field> missing = Sets.difference(ALL, has);
assertHas(flags, has);
assertMissing(flags, missing);
}
@ -117,10 +117,10 @@ public class CommandChangeTest
}
}
private void assertHas(int flags, Set<Fields> missing)
private void assertHas(int flags, Set<Field> 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<Fields> missing)
private void assertMissing(int flags, Set<Field> 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();

1
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

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

2
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

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

2
test/unit/org/apache/cassandra/utils/AccordGenerators.java
View File

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