diff --git a/modules/accord b/modules/accord index 4a8566af7b..8fb29905ee 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 4a8566af7b7de2ddec2c7527d7e2da593f99865f +Subproject commit 8fb29905ee96e089640ff88cccafdf43846b9262 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index f39080bf08..0f174e640f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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 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 bootstrapBeganAt) { if (bootstrapBeganAt != null) diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index bc34a7ad50..6d9744310f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -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; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index eb8c0007f2..310bbeb636 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -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 loadBootstrapBeganAt(int store) { @@ -248,6 +244,34 @@ public class AccordJournal implements IJournal, Shutdownable journal.onFlush(pointer, onFlush); } + @Override + public PersistentField.Persister durableBeforePersister() + { + return new PersistentField.Persister<>() + { + @Override + public AsyncResult persist(DurableBefore addDurableBefore, DurableBefore newDurableBefore) + { + if (isReplay.get()) + return AsyncResults.success(null); + + AsyncResult.Settable 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) diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 34cb57ed5b..0b256574f4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -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 newSafeToRead) { @@ -341,15 +334,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore newBootstrapBeganAt; public NavigableMap newSafeToRead; public RangesForEpoch.Snapshot newRangesForEpoch; diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 87872666bd..e1e88cde05 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -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 historic = Collections.emptyList();} Ref ref = new Ref(); diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java index 1f5088365d..bdf1aa9888 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java @@ -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 map = new TreeMap<>(); for (TxnId txnId : possibleTxns) { diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java index e327c1a5ce..7de9d68602 100644 --- a/src/java/org/apache/cassandra/service/accord/IJournal.java +++ b/src/java/org/apache/cassandra/service/accord/IJournal.java @@ -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 loadBootstrapBeganAt(int commandStoreId); NavigableMap loadSafeToRead(int commandStoreId); CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int commandStoreId); List loadHistoricalTransactions(int store); void appendCommand(int store, SavedCommand.DiffWriter value, Runnable onFlush); + Persister durableBeforePersister(); void persistStoreState(int store, // TODO: this class should not live under ASCS AccordSafeCommandStore.FieldUpdates fieldUpdates, diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java index 6229e8148f..a24cc81e9b 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java @@ -93,23 +93,7 @@ public class AccordJournalCompactionTest IdentityAccumulator rangesForEpochAccumulator = new IdentityAccumulator<>(null); HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator(); - Gen basicRedundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner()); - Gen 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 redundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner()); Gen durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner()); Gen> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner()); Gen 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)); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 3e74728f70..c75690f2e2 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -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 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 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(); } diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java index dd7377ab58..64caa97216 100644 --- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java +++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java @@ -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> bootstrapBeganAtAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY)); final IdentityAccumulator> safeToReadAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY)); final IdentityAccumulator rangesForEpochAccumulator = new IdentityAccumulator<>(null); final HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator(); } + final DurableBeforeAccumulator durableBeforeAccumulator = new DurableBeforeAccumulator(); private final Map 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 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) diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 78708f5538..72581b0cf2 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -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 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> 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 -> { diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index f9f09f5c5c..74ee74a4d1 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -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 -> {