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

Accord: Share DurableBefore between CommandStores

This commit is contained in:
Benedict Elliott Smith 2024-10-03 13:24:03 +01:00 committed by David Capwell
parent 8f7b1c2351
commit 0e1819bbb3
13 changed files with 96 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 4a8566af7b7de2ddec2c7527d7e2da593f99865f
Subproject commit 8fb29905ee96e089640ff88cccafdf43846b9262

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

@ -48,9 +48,8 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Commands;
import accord.local.DurableBefore;
import accord.local.KeyHistory;
import accord.local.NodeTimeService;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
@ -173,7 +172,7 @@ public class AccordCommandStore extends CommandStore
}
public AccordCommandStore(int id,
NodeTimeService time,
NodeCommandStoreService node,
Agent agent,
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
@ -182,7 +181,7 @@ public class AccordCommandStore extends CommandStore
IJournal journal,
CommandStoreExecutor commandStoreExecutor)
{
super(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
super(id, node, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
this.journal = journal;
loggingId = String.format("[%s]", id);
executor = commandStoreExecutor;
@ -219,7 +218,6 @@ public class AccordCommandStore extends CommandStore
this.commandsForRangesLoader = new CommandsForRangesLoader(this);
loadRedundantBefore(journal.loadRedundantBefore(id()));
loadDurableBefore(journal.loadDurableBefore(id()));
loadBootstrapBeganAt(journal.loadBootstrapBeganAt(id()));
loadSafeToRead(journal.loadSafeToRead(id()));
loadRangesForEpoch(journal.loadRangesForEpoch(id()));
@ -230,8 +228,8 @@ public class AccordCommandStore extends CommandStore
static Factory factory(AccordJournal journal, IntFunction<CommandStoreExecutor> executorFactory)
{
return (id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch) ->
new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, executorFactory.apply(id));
return (id, node, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch) ->
new AccordCommandStore(id, node, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, executorFactory.apply(id));
}
public CommandsForRangesLoader diskCommandsForRanges()
@ -417,9 +415,9 @@ public class AccordCommandStore extends CommandStore
return store;
}
NodeTimeService time()
NodeCommandStoreService node()
{
return time;
return node;
}
ProgressLog progressLog()
@ -545,7 +543,7 @@ public class AccordCommandStore extends CommandStore
@VisibleForTesting
public Command loadCommand(TxnId txnId)
{
return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), unsafeGetDurableBefore());
return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore());
}
public interface Loader
@ -592,7 +590,7 @@ public class AccordCommandStore extends CommandStore
Command local = command;
if (local.status() != Truncated && local.status() != Invalidated)
{
Cleanup cleanup = Cleanup.shouldCleanup(local, unsafeGetRedundantBefore(), unsafeGetDurableBefore());
Cleanup cleanup = Cleanup.shouldCleanup(local, unsafeGetRedundantBefore(), durableBefore());
switch (cleanup)
{
case NO:
@ -654,12 +652,6 @@ public class AccordCommandStore extends CommandStore
unsafeSetRedundantBefore(redundantBefore);
}
void loadDurableBefore(DurableBefore durableBefore)
{
if (durableBefore != null)
unsafeSetDurableBefore(durableBefore);
}
void loadBootstrapBeganAt(NavigableMap<TxnId, Ranges> bootstrapBeganAt)
{
if (bootstrapBeganAt != null)

6
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -28,7 +28,7 @@ import accord.api.ProgressLog;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Node;
import accord.local.NodeTimeService;
import accord.local.NodeCommandStoreService;
import accord.local.ShardDistributor;
import accord.primitives.Range;
import accord.topology.Topology;
@ -52,11 +52,11 @@ public class AccordCommandStores extends CommandStores implements CacheSize
private final CommandStoreExecutor[] executors;
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
AccordCommandStores(NodeCommandStoreService node, Agent agent, DataStore store, RandomSource random,
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenerFactory,
AccordJournal journal, CommandStoreExecutor[] executors)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory,
super(node, agent, store, random, shardDistributor, progressLogFactory, listenerFactory,
AccordCommandStore.factory(journal, id -> executors[id % executors.length]));
setCapacity(DatabaseDescriptor.getAccordCacheSizeInMiB() << 20);
this.executors = executors;

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

@ -46,6 +46,9 @@ import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.PersistentField;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.DataInputBuffer;
@ -196,13 +199,6 @@ public class AccordJournal implements IJournal, Shutdownable
return accumulator.get();
}
@Override
public DurableBefore loadDurableBefore(int store)
{
DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, store));
return accumulator.get();
}
@Override
public NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int store)
{
@ -248,6 +244,34 @@ public class AccordJournal implements IJournal, Shutdownable
journal.onFlush(pointer, onFlush);
}
@Override
public PersistentField.Persister<DurableBefore, DurableBefore> durableBeforePersister()
{
return new PersistentField.Persister<>()
{
@Override
public AsyncResult<?> persist(DurableBefore addDurableBefore, DurableBefore newDurableBefore)
{
if (isReplay.get())
return AsyncResults.success(null);
AsyncResult.Settable<Void> result = AsyncResults.settable();
JournalKey key = new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0);
RecordPointer pointer = appendInternal(key, addDurableBefore);
// TODO (required): what happens on failure?
journal.onFlush(pointer, () -> result.setSuccess(null));
return result;
}
@Override
public DurableBefore load()
{
DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0));
return accumulator.get();
}
};
}
@Override
public void persistStoreState(int store, AccordSafeCommandStore.FieldUpdates fieldUpdates, Runnable onFlush)
{
@ -255,8 +279,6 @@ public class AccordJournal implements IJournal, Shutdownable
// TODO: avoid allocating keys
if (fieldUpdates.addRedundantBefore != null)
pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), fieldUpdates.addRedundantBefore);
if (fieldUpdates.addDurableBefore != null)
pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, store), fieldUpdates.addDurableBefore);
if (fieldUpdates.newBootstrapBeganAt != null)
pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), fieldUpdates.newBootstrapBeganAt);
if (fieldUpdates.newSafeToRead != null)

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

@ -35,7 +35,6 @@ import accord.impl.AbstractSafeCommandStore;
import accord.impl.CommandsSummary;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
@ -182,7 +181,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
public NodeTimeService time()
{
// TODO: safe command store should not have arbitrary time
return commandStore.time();
return commandStore.node();
}
@Override
@ -295,12 +294,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
ensureFieldUpdates().newBootstrapBeganAt = newBootstrapBeganAt;
}
@Override
public void upsertDurableBefore(DurableBefore addDurableBefore)
{
ensureFieldUpdates().addDurableBefore = addDurableBefore;
}
@Override
public void setSafeToRead(NavigableMap<Timestamp, Ranges> newSafeToRead)
{
@ -341,15 +334,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return super.redundantBefore();
}
@Override
public DurableBefore durableBefore()
{
if (fieldUpdates != null && fieldUpdates.newDurableBefore != null)
return fieldUpdates.newDurableBefore;
return super.durableBefore();
}
@Override
protected void registerHistoricalTransactions(Deps deps)
{
@ -378,9 +362,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
if (fieldUpdates.newRedundantBefore != null)
super.unsafeSetRedundantBefore(fieldUpdates.newRedundantBefore);
if (fieldUpdates.newDurableBefore != null)
super.unsafeSetDurableBefore(fieldUpdates.newDurableBefore);
if (fieldUpdates.newBootstrapBeganAt != null)
super.setBootstrapBeganAt(fieldUpdates.newBootstrapBeganAt);
@ -394,7 +375,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
public static class FieldUpdates
{
public RedundantBefore addRedundantBefore, newRedundantBefore;
public DurableBefore addDurableBefore, newDurableBefore;
public NavigableMap<TxnId, Ranges> newBootstrapBeganAt;
public NavigableMap<Timestamp, Ranges> newSafeToRead;
public RangesForEpoch.Snapshot newRangesForEpoch;

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

@ -426,6 +426,7 @@ public class AccordService implements IAccordService, Shutdownable
DefaultLocalListeners.Factory::new,
AccordCommandStores.factory(journal),
new AccordInteropFactory(agent, configService),
journal.durableBeforePersister(),
configuration);
this.nodeShutdown = toShutdownable(node);
this.durabilityScheduling = new CoordinateDurabilityScheduling(node);
@ -438,6 +439,7 @@ public class AccordService implements IAccordService, Shutdownable
if (state != State.INIT)
return;
journal.start(node);
node.load();
ClusterMetadataService cms = ClusterMetadataService.instance();
class Ref { List<ClusterMetadata> historic = Collections.emptyList();}
Ref ref = new Ref();

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

@ -189,7 +189,7 @@ public class CommandsForRangesLoader
{
//TODO (now): this logic is kinda duplicate of org.apache.cassandra.service.accord.CommandsForRange.mapReduce
// should figure out if this can be improved... also what is correct?
DurableBefore durableBefore = store.unsafeGetDurableBefore();
DurableBefore durableBefore = store.durableBefore();
NavigableMap<TxnId, Summary> map = new TreeMap<>();
for (TxnId txnId : possibleTxns)
{

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

@ -29,19 +29,20 @@ import accord.primitives.Deps;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.PersistentField.Persister;
public interface IJournal
{
Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore);
RedundantBefore loadRedundantBefore(int commandStoreId);
DurableBefore loadDurableBefore(int commandStoreId);
NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int commandStoreId);
NavigableMap<Timestamp, Ranges> loadSafeToRead(int commandStoreId);
CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int commandStoreId);
List<Deps> loadHistoricalTransactions(int store);
void appendCommand(int store, SavedCommand.DiffWriter value, Runnable onFlush);
Persister<DurableBefore, DurableBefore> durableBeforePersister();
void persistStoreState(int store,
// TODO: this class should not live under ASCS
AccordSafeCommandStore.FieldUpdates fieldUpdates,

31
test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java
View File

@ -93,23 +93,7 @@ public class AccordJournalCompactionTest
IdentityAccumulator<RangesForEpoch.Snapshot> rangesForEpochAccumulator = new IdentityAccumulator<>(null);
HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator();
Gen<RedundantBefore> basicRedundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner());
Gen<RedundantBefore> redundantBeforeGen = rs -> {
// TODO: find a better way to generate consecutive redundant befores
while (true)
{
RedundantBefore next = basicRedundantBeforeGen.next(rs);
try
{
RedundantBefore.merge(redundantBeforeAccumulator.get(), next);
return next;
}
catch (Throwable t)
{
// retry;
}
}
};
Gen<RedundantBefore> redundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner());
Gen<DurableBefore> durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner());
Gen<NavigableMap<Timestamp, Ranges>> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner());
Gen<RangesForEpoch.Snapshot> rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner());
@ -137,21 +121,24 @@ public class AccordJournalCompactionTest
RandomSource rs = new DefaultRandom();
int count = 1_000;
// RedundantBefore redundantBefore = RedundantBefore.EMPTY;
for (int i = 0; i <= count; i++)
{
timestamp = timestamp.next();
AccordSafeCommandStore.FieldUpdates updates = new AccordSafeCommandStore.FieldUpdates();
updates.addDurableBefore = durableBeforeGen.next(rs);
DurableBefore addDurableBefore = durableBeforeGen.next(rs);
// TODO: improve redundant before generator and re-enable
// updates.redundantBefore = redundantBeforeGen.next(rs);
// updates.addRedundantBefore = redundantBeforeGen.next(rs);
// updates.newRedundantBefore = redundantBefore = RedundantBefore.merge(redundantBefore, updates.addRedundantBefore);
updates.newSafeToRead = safeToReadGen.next(rs);
updates.newRangesForEpoch = rangesForEpochGen.next(rs);
updates.addHistoricalTransactions = historicalTransactionsGen.next(rs);
journal.durableBeforePersister().persist(addDurableBefore, null);
journal.persistStoreState(1, updates, null);
redundantBeforeAccumulator.update(updates.addRedundantBefore);
durableBeforeAccumulator.update(updates.addDurableBefore);
redundantBeforeAccumulator.update(updates.newRedundantBefore);
durableBeforeAccumulator.update(addDurableBefore);
if (updates.newBootstrapBeganAt != null)
bootstrapBeganAtAccumulator.update(updates.newBootstrapBeganAt);
safeToReadAccumulator.update(updates.newSafeToRead);
@ -165,7 +152,7 @@ public class AccordJournalCompactionTest
}
// Assert.assertEquals(redundantBeforeAccumulator.get(), journal.loadRedundantBefore(1));
Assert.assertEquals(durableBeforeAccumulator.get(), journal.loadDurableBefore(1));
Assert.assertEquals(durableBeforeAccumulator.get(), journal.durableBeforePersister().load());
Assert.assertEquals(bootstrapBeganAtAccumulator.get(), journal.loadBootstrapBeganAt(1));
Assert.assertEquals(safeToReadAccumulator.get(), journal.loadSafeToRead(1));
Assert.assertEquals(rangesForEpochAccumulator.get(), journal.loadRangesForEpoch(1));

11
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -46,8 +46,10 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.CommonAttributes;
import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.NodeCommandStoreService;
import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@ -364,16 +366,16 @@ public class AccordTestUtils
TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
TokenRange range = TokenRange.fullRange(metadata.id);
Node.Id node = new Id(1);
NodeTimeService time = new NodeTimeService()
NodeCommandStoreService time = new NodeCommandStoreService()
{
private ToLongFunction<TimeUnit> elapsed = NodeTimeService.elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, this::now);
@Override public Id id() { return node;}
@Override public DurableBefore durableBefore() { return DurableBefore.EMPTY; }
@Override public long epoch() {return 1; }
@Override public long now() {return now.getAsLong(); }
@Override public Timestamp uniqueNow(Timestamp atLeast) { return Timestamp.fromValues(1, now.getAsLong(), node); }
@Override
public long elapsed(TimeUnit timeUnit) { return elapsed.applyAsLong(timeUnit); }
@Override public long elapsed(TimeUnit timeUnit) { return elapsed.applyAsLong(timeUnit); }
};
SingleEpochRanges holder = new SingleEpochRanges(Ranges.of(range));
@ -386,10 +388,11 @@ public class AccordTestUtils
public static AccordCommandStore createAccordCommandStore(
Node.Id node, LongSupplier now, Topology topology, ExecutorPlus loadExecutor, ExecutorPlus saveExecutor)
{
NodeTimeService time = new NodeTimeService()
NodeCommandStoreService time = new NodeCommandStoreService()
{
private ToLongFunction<TimeUnit> elapsed = NodeTimeService.elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, this::now);
@Override public DurableBefore durableBefore() { return DurableBefore.EMPTY; }
@Override public Id id() { return node;}
@Override public long epoch() {return 1; }
@Override public long now() {return now.getAsLong(); }

25
test/unit/org/apache/cassandra/service/accord/MockJournal.java
View File

@ -47,6 +47,9 @@ import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.Invariants;
import accord.utils.PersistentField.Persister;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsAccumulator;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator;
@ -60,13 +63,13 @@ public class MockJournal implements IJournal
private static class FieldUpdates
{
final RedundantBeforeAccumulator redundantBeforeAccumulator = new RedundantBeforeAccumulator();
final DurableBeforeAccumulator durableBeforeAccumulator = new DurableBeforeAccumulator();
final IdentityAccumulator<NavigableMap<TxnId, Ranges>> bootstrapBeganAtAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY));
final IdentityAccumulator<NavigableMap<Timestamp, Ranges>> safeToReadAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY));
final IdentityAccumulator<CommandStores.RangesForEpoch.Snapshot> rangesForEpochAccumulator = new IdentityAccumulator<>(null);
final HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator();
}
final DurableBeforeAccumulator durableBeforeAccumulator = new DurableBeforeAccumulator();
private final Map<Integer, FieldUpdates> fieldUpdates = new HashMap<>();
@Override
public Command loadCommand(int store, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
@ -85,9 +88,23 @@ public class MockJournal implements IJournal
}
@Override
public DurableBefore loadDurableBefore(int store)
public Persister<DurableBefore, DurableBefore> durableBeforePersister()
{
return fieldUpdates(store).durableBeforeAccumulator.get();
return new Persister<>()
{
@Override
public AsyncResult<?> persist(DurableBefore addDurableBefore, DurableBefore newDurableBefore)
{
durableBeforeAccumulator.update(addDurableBefore);
return AsyncResults.success(null);
}
@Override
public DurableBefore load()
{
return durableBeforeAccumulator.get();
}
};
}
@Override
@ -139,8 +156,6 @@ public class MockJournal implements IJournal
FieldUpdates updates = fieldUpdates(store);
if (fieldUpdates.addRedundantBefore != null)
updates.redundantBeforeAccumulator.update(fieldUpdates.addRedundantBefore);
if (fieldUpdates.addDurableBefore != null)
updates.durableBeforeAccumulator.update(fieldUpdates.addDurableBefore);
if (fieldUpdates.newBootstrapBeganAt != null)
updates.bootstrapBeganAtAccumulator.update(fieldUpdates.newBootstrapBeganAt);
if (fieldUpdates.newSafeToRead != null)

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

@ -38,7 +38,9 @@ import accord.impl.TestAgent;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.NodeCommandStoreService;
import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@ -92,7 +94,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
private final CommandStore.EpochUpdateHolder updateHolder;
private final BooleanSupplier shouldEvict, shouldFlush, shouldCompact;
public final NodeTimeService timeService;
public final NodeCommandStoreService storeService;
public final AccordCommandStore store;
public final Node.Id nodeId;
public final Topology topology;
@ -114,10 +116,12 @@ public class SimulatedAccordCommandStore implements AutoCloseable
this.updateHolder = new CommandStore.EpochUpdateHolder();
this.nodeId = AccordTopology.tcmIdToAccord(ClusterMetadata.currentNullable().myNodeId());
this.timeService = new NodeTimeService()
this.storeService = new NodeCommandStoreService()
{
private final ToLongFunction<TimeUnit> elapsed = NodeTimeService.elapsedWrapperFromNonMonotonicSource(TimeUnit.NANOSECONDS, this::now);
@Override public DurableBefore durableBefore() { return DurableBefore.EMPTY; }
@Override
public Node.Id id()
{
@ -155,7 +159,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
AccordStateCache stateCache = new AccordStateCache(Stage.READ.executor(), Stage.MUTATION.executor(), 8 << 20, new AccordStateCacheMetrics("test"));
this.journal = new MockJournal();
this.store = new AccordCommandStore(0,
timeService,
storeService,
new TestAgent.RethrowAgent()
{
@Override
@ -222,7 +226,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
public TxnId nextTxnId(Txn.Kind kind, Routable.Domain domain)
{
return new TxnId(timeService.epoch(), timeService.now(), kind, domain, nodeId);
return new TxnId(storeService.epoch(), storeService.now(), kind, domain, nodeId);
}
public void maybeCacheEvict(Unseekables<?> keysOrRanges)
@ -362,7 +366,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
public Pair<TxnId, AsyncResult<BeginRecovery.RecoverOk>> enqueueBeginRecovery(Txn txn, FullRoute<?> route)
{
TxnId txnId = nextTxnId(txn.kind(), txn.keys().domain());
Ballot ballot = Ballot.fromValues(timeService.epoch(), timeService.now(), nodeId);
Ballot ballot = Ballot.fromValues(storeService.epoch(), storeService.now(), nodeId);
BeginRecovery br = new BeginRecovery(nodeId, topologies, txnId, null, txn, route, ballot);
return Pair.create(txnId, processAsync(br, safe -> {

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

@ -268,7 +268,7 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
return success;
});
var delay = preAcceptAsync.flatMap(ignore -> AsyncChains.ofCallable(instance.unorderedScheduled, () -> {
Ballot ballot = Ballot.fromValues(instance.timeService.epoch(), instance.timeService.now(), nodeId);
Ballot ballot = Ballot.fromValues(instance.storeService.epoch(), instance.storeService.now(), nodeId);
return new BeginRecovery(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, null, txn, route, ballot);
}));
var recoverAsync = delay.flatMap(br -> instance.processAsync(br, safe -> {