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

Refactor RedundantStatus to encode vector of states that can be merged independently 

Also fix:
 - Truncate command on first access, without participants
 - Use Ballot.ZERO when invoking CFK.insertOutOfRange where appropriate
 - Don't supply a command's own route to ProgressLog.waiting to ensure new keys are incorporated
 - Ensure progress in CommandsForKey by setting vestigial commands to ERASED
 - Add any missing owned keys to StoreParticipants.route to ensure fetch can make progress
 - Recovery must wait for earlier not-accepted transactions if either has the privileged coordinator optimisation
 - Inclusive SyncPoint used incorrect topologies for propose phase
 - Barrier must not register local listener without up-to-date topology information
 - Stop home shard truncating a TxnId to vestigial rather than Invalidated so other shards can make progress
Also improve:
 - Validate commands are constructed with non-empty participants
 - Remove some unnecessary synchronized keywords
 - Clear ok messages on PreAccept and Accept to free up memory
 - Introduce TxnId.Cardinality flag so we can optimise single key queries
 - Update CommandsForKey serialization to better handle larger flag space
 - Configurable which Txn.Kind can result in a CommandStore being marked stale
 - Process DefaultProgressLog queue synchronously when relevant state is resident in memory
 - Remove defunct CollectMaxApplied version of ListStore bootstrap
 - Standardise linearizability violation reporting
 - Improve CommandStore.execute method naming to reduce chance of misuse
 - Prune and address some comments

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20282
This commit is contained in:
Benedict Elliott Smith 2025-02-01 17:42:21 +00:00 committed by David Capwell
parent c4a5971af3
commit af7dfcc327
15 changed files with 109 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 78ab7eef904ef549d0d7a34332b83d6110e0762d
Subproject commit cd7f49564a5ad053286453d10f8cd46b8d870c4f

8
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -472,11 +472,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace
ds.row(storeId, decompose(start), decompose(end))
.column("start_ownership_epoch", entry.startOwnershipEpoch)
.column("end_ownership_epoch", entry.endOwnershipEpoch)
.column("locally_applied_or_invalidated_before", entry.locallyAppliedOrInvalidatedBefore.toString())
.column("locally_decided_and_applied_or_invalidated_before", entry.locallyDecidedAndAppliedOrInvalidatedBefore.toString())
.column("shard_applied_or_invalidated_before", entry.shardAppliedOrInvalidatedBefore.toString())
.column("locally_applied_before", entry.locallyAppliedBefore.toString())
.column("locally_decided_and_applied_before", entry.locallyDecidedAndAppliedBefore.toString())
.column("shard_applied_before", entry.shardAppliedBefore.toString())
.column("gc_before", entry.gcBefore.toString())
.column("shard_only_applied_or_invalidated_before", entry.shardOnlyAppliedOrInvalidatedBefore.toString())
.column("shard_only_applied_before", entry.shardOnlyAppliedBefore.toString())
.column("bootstrapped_at", entry.bootstrappedAt.toString())
.column("stale_until_at_least", entry.staleUntilAtLeast != null ? entry.staleUntilAtLeast.toString() : null);
return ds;

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

@ -309,7 +309,7 @@ public class AccordCommandStore extends CommandStore
return lastSystemTimestampMicros;
}
@Override
public <T> AsyncChain<T> submit(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
public <T> AsyncChain<T> build(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
{
return AccordTask.create(this, loadCtx, function).chain();
}
@ -336,7 +336,7 @@ public class AccordCommandStore extends CommandStore
}
@Override
public AsyncChain<Void> execute(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
public AsyncChain<Void> build(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
{
return AccordTask.create(this, preLoadContext, consumer).chain();
}

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

@ -71,13 +71,13 @@ public class AccordMessageSink implements MessageSink
public static final class AccordMessageType extends MessageType
{
public static final AccordMessageType INTEROP_READ_REQ = remote("INTEROP_READ_REQ", false);
public static final AccordMessageType INTEROP_READ_RSP = remote("INTEROP_READ_RSP", false);
public static final AccordMessageType INTEROP_STABLE_THEN_READ_REQ = remote("INTEROP_STABLE_THEN_READ_REQ", false);
public static final AccordMessageType INTEROP_READ_REPAIR_REQ = remote("INTEROP_READ_REPAIR_REQ", false);
public static final AccordMessageType INTEROP_READ_REPAIR_RSP = remote("INTEROP_READ_REPAIR_RSP", false);
public static final AccordMessageType INTEROP_APPLY_MINIMAL_REQ = remote("INTEROP_APPLY_MINIMAL_REQ", true );
public static final AccordMessageType INTEROP_APPLY_MAXIMAL_REQ = remote("INTEROP_APPLY_MAXIMAL_REQ", true );
public static final AccordMessageType INTEROP_READ_REQ = remote("INTEROP_READ_REQ");
public static final AccordMessageType INTEROP_READ_RSP = remote("INTEROP_READ_RSP");
public static final AccordMessageType INTEROP_STABLE_THEN_READ_REQ = remote("INTEROP_STABLE_THEN_READ_REQ");
public static final AccordMessageType INTEROP_READ_REPAIR_REQ = remote("INTEROP_READ_REPAIR_REQ");
public static final AccordMessageType INTEROP_READ_REPAIR_RSP = remote("INTEROP_READ_REPAIR_RSP");
public static final AccordMessageType INTEROP_APPLY_MINIMAL_REQ = remote("INTEROP_APPLY_MINIMAL_REQ");
public static final AccordMessageType INTEROP_APPLY_MAXIMAL_REQ = remote("INTEROP_APPLY_MAXIMAL_REQ");
public static final List<MessageType> values;
@ -101,14 +101,14 @@ public class AccordMessageSink implements MessageSink
values = builder.build();
}
protected static AccordMessageType remote(String name, boolean hasSideEffects)
protected static AccordMessageType remote(String name)
{
return new AccordMessageType(name, REMOTE, hasSideEffects);
return new AccordMessageType(name, REMOTE);
}
private AccordMessageType(String name, MessageType.Kind kind, boolean hasSideEffects)
private AccordMessageType(String name, MessageType.Kind kind)
{
super(name, kind, hasSideEffects);
super(name, kind);
}
}

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

@ -44,6 +44,7 @@ import accord.primitives.PartialDeps;
import accord.primitives.PartialKeyRoute;
import accord.primitives.PartialRangeRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
import accord.primitives.Range;
import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
@ -290,8 +291,9 @@ public class AccordObjectSizes
private static ICommand attrs(boolean hasDeps, boolean hasTxn, boolean executes)
{
FullKeyRoute route = new FullKeyRoute(EMPTY_KEY, new RoutingKey[]{ EMPTY_KEY });
Participants<?> empty = route.slice(0, 0);
ICommand.Builder builder = new ICommand.Builder(EMPTY_TXNID)
.setParticipants(StoreParticipants.empty(EMPTY_TXNID, route, !executes))
.setParticipants(StoreParticipants.create(route, empty, executes ? empty : null, empty, route))
.durability(Status.Durability.NotDurable)
.executeAt(EMPTY_TXNID)
.promised(Ballot.ZERO);
@ -316,8 +318,8 @@ public class AccordObjectSizes
final static long ACCEPTED = measure(Command.Accepted.accepted(attrs(true, false, false), SaveStatus.AcceptedMedium));
final static long COMMITTED = measure(Command.Committed.committed(attrs(true, true, false), SaveStatus.Committed));
final static long EXECUTED = measure(Command.Executed.executed(attrs(true, true, true), SaveStatus.Applied));
final static long TRUNCATED = measure(Command.Truncated.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
final static long INVALIDATED = measure(Command.Truncated.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID)));
final static long TRUNCATED = measure(Command.Truncated.truncated(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
final static long INVALIDATED = measure(Command.Truncated.invalidated(EMPTY_TXNID, attrs(false, false, false).participants()));
private static long emptySize(Command command)
{
@ -353,7 +355,7 @@ public class AccordObjectSizes
case Applied:
return EXECUTED;
case TruncatedApply:
case TruncatedApplyWithDeps:
case TruncatedUnapplied:
case TruncatedApplyWithOutcome:
case Vestigial:
case Erased:

3
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -173,6 +173,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.primitives.TxnId.Cardinality.cardinality;
import static accord.utils.Invariants.require;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ -818,7 +819,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public @Nonnull AsyncTxnResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, @Nonnull Dispatcher.RequestTime requestTime)
{
TxnId txnId = node.nextTxnId(txn.kind(), txn.keys().domain());
TxnId txnId = node.nextTxnId(txn.kind(), txn.keys().domain(), cardinality(txn.keys()));
ClientRequestMetrics sharedMetrics;
AccordClientRequestMetrics metrics;
if (txn.isWrite())

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

@ -20,6 +20,8 @@ package org.apache.cassandra.service.accord.api;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -35,6 +37,7 @@ import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.messages.ReplyContext;
import accord.primitives.Keys;
import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Seekables;
@ -59,7 +62,6 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.Write;
import static accord.utils.Invariants.illegalState;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ -259,9 +261,8 @@ public class AccordAgent implements Agent
}
@Override
public void onViolation(String message)
public void onViolation(String message, Participants<?> participants, @Nullable TxnId notWitnessed, @Nullable Timestamp notWitnessedExecuteAt, @Nullable TxnId by, @Nullable Timestamp byEexecuteAt)
{
try { throw illegalState(message); }
catch (Throwable t) { logger.error("Consistency violation", t); }
logger.error(message);
}
}

63
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
View File

@ -94,7 +94,12 @@ public class CommandSerializers
long epoch = txnId.epoch();
if((flags & HAS_EPOCH) != 0)
epoch += in.readUnsignedVInt();
{
long delta = in.readUnsignedVInt();
if (delta == 0)
return Timestamp.NONE;
epoch += delta - 1;
}
long hlc = txnId.hlc() + in.readUnsignedVInt();
Node.Id node = new Node.Id(in.readUnsignedVInt32());
@ -108,8 +113,8 @@ public class CommandSerializers
int flags = in.readUnsignedVInt32();
if ((flags & 1) != 0)
{
if ((flags & HAS_EPOCH) != 0)
in.readUnsignedVInt();
if ((flags & HAS_EPOCH) != 0 && in.readUnsignedVInt() == 0)
return;
in.readUnsignedVInt();
in.readUnsignedVInt32();
if ((flags & HAS_UNIQUE_HLC) != 0)
@ -124,7 +129,14 @@ public class CommandSerializers
if ((flags & 1) != 0)
{
if ((flags & HAS_EPOCH) != 0)
out.writeUnsignedVInt(executeAt.epoch() - txnId.epoch());
{
if (executeAt.equals(Timestamp.NONE))
{
out.writeUnsignedVInt(0L);
return;
}
out.writeUnsignedVInt(1 + executeAt.epoch() - txnId.epoch());
}
out.writeUnsignedVInt(executeAt.hlc() - txnId.hlc());
out.writeUnsignedVInt32(executeAt.node.id);
if ((flags & HAS_UNIQUE_HLC) != 0)
@ -152,7 +164,12 @@ public class CommandSerializers
if ((flags & 1) != 0)
{
if ((flags & HAS_EPOCH) != 0)
{
if (executeAt.equals(Timestamp.NONE))
return size + TypeSizes.sizeofUnsignedVInt(0L);
size += TypeSizes.sizeofUnsignedVInt(executeAt.epoch() - txnId.epoch());
}
size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc() - txnId.hlc());
size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id);
if ((flags & HAS_UNIQUE_HLC) != 0)
@ -176,10 +193,13 @@ public class CommandSerializers
int flags = in.readUnsignedVInt32();
if (nullable)
{
if ((flags & 1) != 0) return null;
flags >>>= 1;
if (flags == 0) return null;
flags--;
}
long epoch = in.readUnsignedVInt();
if (epoch-- == 0)
return Timestamp.NONE;
long hlc = in.readUnsignedVInt();
Node.Id node = new Node.Id(in.readUnsignedVInt32());
if ((flags & HAS_UNIQUE_HLC) == 0)
@ -206,11 +226,12 @@ public class CommandSerializers
int flags = in.readUnsignedVInt32();
if (nullable)
{
if ((flags & 1) != 0)
if (flags == 0)
return;
flags >>>= 1;
flags--;
}
in.readUnsignedVInt();
if (0 == in.readUnsignedVInt())
return;
in.readUnsignedVInt();
in.readUnsignedVInt32();
if ((flags & HAS_UNIQUE_HLC) != 0)
@ -235,9 +256,13 @@ public class CommandSerializers
{
Invariants.require(nullable);
}
else if (executeAt.equals(Timestamp.NONE))
{
out.writeUnsignedVInt(0L);
}
else
{
out.writeUnsignedVInt(executeAt.epoch());
out.writeUnsignedVInt(1 + executeAt.epoch());
out.writeUnsignedVInt(executeAt.hlc());
out.writeUnsignedVInt32(executeAt.node.id);
if (executeAt.hasDistinctHlcAndUniqueHlc())
@ -264,11 +289,15 @@ public class CommandSerializers
Invariants.require(nullable);
return size;
}
size += TypeSizes.sizeofUnsignedVInt(executeAt.epoch());
size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc());
size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id);
if (executeAt.hasDistinctHlcAndUniqueHlc())
size += TypeSizes.sizeofUnsignedVInt(executeAt.uniqueHlc() - executeAt.hlc());
if (executeAt.equals(Timestamp.NONE)) size += TypeSizes.sizeofUnsignedVInt(0);
else
{
size += TypeSizes.sizeofUnsignedVInt(1 + executeAt.epoch());
size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc());
size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id);
if (executeAt.hasDistinctHlcAndUniqueHlc())
size += TypeSizes.sizeofUnsignedVInt(executeAt.uniqueHlc() - executeAt.hlc());
}
return size;
}
@ -277,7 +306,7 @@ public class CommandSerializers
if (executeAt == null)
{
Invariants.require(nullable);
return 1;
return 0;
}
int flags = executeAt.flags() << 2;
@ -286,7 +315,7 @@ public class CommandSerializers
if (executeAt.hasDistinctHlcAndUniqueHlc())
flags |= HAS_UNIQUE_HLC;
if (nullable)
flags <<= 1;
flags++;
return flags;
}
}

20
src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java
View File

@ -142,11 +142,11 @@ public class CommandStoreSerializers
out.writeUnsignedVInt(t.startOwnershipEpoch);
if (t.endOwnershipEpoch == Long.MAX_VALUE) out.writeUnsignedVInt(0L);
else out.writeUnsignedVInt(1 + t.endOwnershipEpoch - t.startOwnershipEpoch);
CommandSerializers.txnId.serialize(t.locallyWitnessedOrInvalidatedBefore, out, version);
CommandSerializers.txnId.serialize(t.locallyAppliedOrInvalidatedBefore, out, version);
CommandSerializers.txnId.serialize(t.locallyDecidedAndAppliedOrInvalidatedBefore, out, version);
CommandSerializers.txnId.serialize(t.shardOnlyAppliedOrInvalidatedBefore, out, version);
CommandSerializers.txnId.serialize(t.shardAppliedOrInvalidatedBefore, out, version);
CommandSerializers.txnId.serialize(t.locallyWitnessedBefore, out, version);
CommandSerializers.txnId.serialize(t.locallyAppliedBefore, out, version);
CommandSerializers.txnId.serialize(t.locallyDecidedAndAppliedBefore, out, version);
CommandSerializers.txnId.serialize(t.shardOnlyAppliedBefore, out, version);
CommandSerializers.txnId.serialize(t.shardAppliedBefore, out, version);
CommandSerializers.txnId.serialize(t.gcBefore, out, version);
CommandSerializers.txnId.serialize(t.bootstrappedAt, out, version);
CommandSerializers.nullableTimestamp.serialize(t.staleUntilAtLeast, out, version);
@ -177,11 +177,11 @@ public class CommandStoreSerializers
long size = KeySerializers.range.serializedSize(t.range, version);
size += TypeSizes.sizeofUnsignedVInt(t.startOwnershipEpoch);
size += TypeSizes.sizeofUnsignedVInt(t.endOwnershipEpoch == Long.MAX_VALUE ? 0 : 1 + t.endOwnershipEpoch - t.startOwnershipEpoch);
size += CommandSerializers.txnId.serializedSize(t.locallyWitnessedOrInvalidatedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.locallyAppliedOrInvalidatedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.locallyDecidedAndAppliedOrInvalidatedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.shardOnlyAppliedOrInvalidatedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.shardAppliedOrInvalidatedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.locallyWitnessedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.locallyAppliedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.locallyDecidedAndAppliedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.shardOnlyAppliedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.shardAppliedBefore, version);
size += CommandSerializers.txnId.serializedSize(t.gcBefore, version);
size += CommandSerializers.txnId.serializedSize(t.bootstrappedAt, version);
size += CommandSerializers.nullableTimestamp.serializedSize(t.staleUntilAtLeast, version);

4
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -53,7 +53,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.Condition;
import org.awaitility.Awaitility;
import static accord.primitives.TxnId.FastPath.UNOPTIMISED;
import static accord.primitives.TxnId.FastPath.Unoptimised;
import static org.apache.cassandra.Util.spinUntilSuccess;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
@ -136,7 +136,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
public void blocked() throws ExecutionException, InterruptedException
{
ProtocolModifiers.Toggles.setPermitLocalExecution(false);
ProtocolModifiers.Toggles.setPermittedFastPaths(new TxnId.FastPaths(UNOPTIMISED));
ProtocolModifiers.Toggles.setPermittedFastPaths(new TxnId.FastPaths(Unoptimised));
AccordMsgFilter filter = new AccordMsgFilter();
MessagingService.instance().outboundSink.add(filter);
try

4
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
View File

@ -634,8 +634,8 @@ public class RouteIndexTest extends CQLTester.InMemory
Int2ObjectHashMap<DurableBefore> durableBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<CommandStores.RangesForEpoch> ranges = new Int2ObjectHashMap<>();
RedundantBefore redundantBefore = Mockito.spy(RedundantBefore.EMPTY);
Mockito.doReturn(RedundantStatus.LIVE).when(redundantBefore).status(Mockito.any(), Mockito.any(), (Participants<?>) Mockito.any());
Mockito.doReturn(RedundantStatus.LIVE).when(redundantBefore).status(Mockito.any(), Mockito.any(), (RoutingKey) Mockito.any());
Mockito.doReturn(RedundantStatus.NONE).when(redundantBefore).status(Mockito.any(), Mockito.any(), (Participants<?>) Mockito.any());
Mockito.doReturn(RedundantStatus.NONE).when(redundantBefore).status(Mockito.any(), Mockito.any(), (RoutingKey) Mockito.any());
for (int i = 0; i < MAX_STORES; i++)
{
redundantBefores.put(i, redundantBefore);

8
test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java
View File

@ -30,6 +30,8 @@ import accord.api.Journal;
import accord.local.Command;
import accord.local.StoreParticipants;
import accord.primitives.Ballot;
import accord.primitives.Participants;
import accord.primitives.RoutingKeys;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.TxnId;
@ -39,13 +41,16 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.TestParams;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.StorageCompatibilityMode;
@ -82,7 +87,8 @@ public class AccordJournalOrderTest
TxnId txnId = randomSource.nextBoolean() ? id1 : id2;
JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, randomSource.nextInt(5));
res.compute(key, (k, prev) -> prev == null ? 1 : prev + 1);
Command command = Command.NotDefined.notDefined(txnId, SaveStatus.NotDefined, Status.Durability.NotDurable, StoreParticipants.empty(txnId), Ballot.ZERO);
Participants<?> participants = RoutingKeys.of(new AccordRoutingKey.TokenKey(TableId.generate(), new Murmur3Partitioner.LongToken(1)));
Command command = Command.NotDefined.notDefined(txnId, SaveStatus.NotDefined, Status.Durability.NotDurable, StoreParticipants.create(null, participants, null, participants, participants), Ballot.ZERO);
accordJournal.saveCommand(key.commandStoreId,
new Journal.CommandUpdate(null, command),
() -> {});

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

@ -95,7 +95,6 @@ public class CommandChangeTest
SoftAssertions checks = new SoftAssertions();
for (SaveStatus saveStatus : SaveStatus.values())
{
if (saveStatus == SaveStatus.TruncatedApplyWithDeps) continue;
out.clear();
Command orig = cmdBuilder.build(saveStatus);

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

@ -476,7 +476,7 @@ public class CommandsForKeySerializerTest
}
// TODO (expected): we currently don't explore TruncatedApply statuses because we don't transition through all phases and therefore don't adopt the Applied status
Choices<SaveStatus> saveStatusChoices = Choices.uniform(EnumSet.complementOf(EnumSet.of(SaveStatus.TruncatedApply, SaveStatus.TruncatedApplyWithOutcome, SaveStatus.TruncatedApplyWithDeps)).toArray(SaveStatus[]::new));
Choices<SaveStatus> saveStatusChoices = Choices.uniform(EnumSet.complementOf(EnumSet.of(SaveStatus.TruncatedApply, SaveStatus.TruncatedUnapplied, SaveStatus.TruncatedApplyWithOutcome)).toArray(SaveStatus[]::new));
Supplier<SaveStatus> saveStatusSupplier = () -> {
SaveStatus result = saveStatusChoices.choose(source);
while (result.is(Status.Truncated)) // we don't currently process truncations
@ -559,7 +559,7 @@ public class CommandsForKeySerializerTest
public void test()
{
var tableGen = AccordGenerators.fromQT(CassandraGenerators.TABLE_ID_GEN);
var txnIdGen = AccordGens.txnIds(rs -> rs.nextLong(0, 100), rs -> rs.nextLong(100), rs -> rs.nextInt(10));
var txnIdGen = AccordGens.txnIds((Gen.LongGen) rs -> rs.nextLong(0, 100), rs -> rs.nextLong(100), rs -> rs.nextInt(10));
qt().check(rs -> {
TableId table = tableGen.next(rs);
TokenKey pk = new TokenKey(table, new Murmur3Partitioner.LongToken(rs.nextLong()));
@ -647,8 +647,8 @@ public class CommandsForKeySerializerTest
@Override public boolean inStore() { return true; }
@Override public Journal.Loader loader() { throw new UnsupportedOperationException(); }
@Override public Agent agent() { return this; }
@Override public AsyncChain<Void> execute(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { return null; }
@Override public <T> AsyncChain<T> submit(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public AsyncChain<Void> build(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { return null; }
@Override public <T> AsyncChain<T> build(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public void shutdown() { }
@Override protected void registerTransitive(SafeCommandStore safeStore, RangeDeps deps){ }
@Override public <T> AsyncChain<T> submit(Callable<T> task) { throw new UnsupportedOperationException(); }

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

@ -33,6 +33,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedMap;
import accord.local.Command;
import accord.local.Command.Truncated;
import accord.local.ICommand;
import accord.local.DurableBefore;
import accord.local.RedundantBefore;
@ -245,8 +246,6 @@ public class AccordGenerators
switch (saveStatus)
{
default: throw new AssertionError("Unhandled saveStatus: " + saveStatus);
case TruncatedApplyWithDeps:
throw new IllegalArgumentException("TruncatedApplyWithDeps is not a valid state for a Command to be in, its for FetchData");
case Uninitialised:
case NotDefined:
return Command.NotDefined.notDefined(attributes(saveStatus), Ballot.ZERO);
@ -285,17 +284,18 @@ public class AccordGenerators
return Command.Executed.executed(attributes(saveStatus), saveStatus);
case TruncatedApply:
if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null, txnId);
else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null);
case TruncatedUnapplied:
if (txnId.kind().awaitsOnlyDeps()) return Truncated.truncated(attributes(saveStatus), saveStatus, executeAt, null, null, txnId);
else return Truncated.truncated(attributes(saveStatus), saveStatus, executeAt, null, null);
case TruncatedApplyWithOutcome:
if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges,new TxnWrite(Collections.emptyList(), true)) : null, new TxnData(), txnId);
else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)) : null, new TxnData());
if (txnId.kind().awaitsOnlyDeps()) return Truncated.truncated(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges,new TxnWrite(Collections.emptyList(), true)) : null, new TxnData(), txnId);
else return Truncated.truncated(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)) : null, new TxnData());
case Erased:
case Vestigial:
case Invalidated:
return Command.Truncated.invalidated(txnId, attributes(saveStatus).participants());
return Truncated.invalidated(txnId, attributes(saveStatus).participants());
}
}
}