You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/10/15 10:36:36 UTC
(cassandra-accord) branch trunk updated: Support loading historical transactions on new topology
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
The following commit(s) were added to refs/heads/trunk by this push:
new d914ee69 Support loading historical transactions on new topology
d914ee69 is described below
commit d914ee69816ebfdf88b2120ff1d8e0bc16edecbc
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Tue Oct 15 11:36:28 2024 +0100
Support loading historical transactions on new topology
---
.../main/java/accord/api/ConfigurationService.java | 2 +-
.../accord/impl/AbstractConfigurationService.java | 12 +++---
.../java/accord/impl/InMemoryCommandStore.java | 2 +-
.../main/java/accord/impl/MajorityDepsFetcher.java | 46 ++++++++++++++++------
.../src/main/java/accord/local/CommandStore.java | 35 +++++++++++-----
.../src/main/java/accord/local/CommandStores.java | 8 ++--
accord-core/src/main/java/accord/local/Node.java | 10 ++---
.../main/java/accord/local/SafeCommandStore.java | 5 ++-
.../impl/AbstractConfigurationServiceTest.java | 2 +-
.../test/java/accord/impl/RemoteListenersTest.java | 5 ++-
.../accord/impl/basic/DelayedCommandStores.java | 4 +-
.../test/java/accord/impl/mock/MockCluster.java | 2 +-
.../accord/impl/mock/MockConfigurationService.java | 2 +-
.../java/accord/local/ImmutableCommandTest.java | 2 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 2 +-
15 files changed, 89 insertions(+), 50 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/ConfigurationService.java b/accord-core/src/main/java/accord/api/ConfigurationService.java
index 5d62c709..b7e7ee83 100644
--- a/accord-core/src/main/java/accord/api/ConfigurationService.java
+++ b/accord-core/src/main/java/accord/api/ConfigurationService.java
@@ -115,7 +115,7 @@ public interface ConfigurationService
*
* TODO (desired): document what this Future represents, or maybe refactor it away - only used for testing
*/
- AsyncResult<Void> onTopologyUpdate(Topology topology, boolean startSync);
+ AsyncResult<Void> onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync);
/**
* Called when accord data associated with a superseded epoch has been sync'd from previous replicas.
diff --git a/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java b/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
index 43b9a658..bb70b14a 100644
--- a/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
+++ b/accord-core/src/main/java/accord/impl/AbstractConfigurationService.java
@@ -267,7 +267,7 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
protected void topologyUpdatePreListenerNotify(Topology topology) {}
protected void topologyUpdatePostListenerNotify(Topology topology) {}
- public synchronized void reportTopology(Topology topology, boolean startSync)
+ public synchronized void reportTopology(Topology topology, boolean isLoad, boolean startSync)
{
long lastReceived = epochs.lastReceived;
if (topology.epoch() <= lastReceived)
@@ -276,19 +276,19 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
if (lastReceived > 0 && topology.epoch() > lastReceived + 1)
{
fetchTopologyForEpoch(lastReceived + 1);
- epochs.receiveFuture(lastReceived + 1).addCallback(() -> reportTopology(topology, startSync));
+ epochs.receiveFuture(lastReceived + 1).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
}
long lastAcked = epochs.lastAcknowledged;
if (lastAcked == 0 && lastReceived > 0)
{
- epochs.acknowledgeFuture(epochs.minEpoch()).addCallback(() -> reportTopology(topology, startSync));
+ epochs.acknowledgeFuture(epochs.minEpoch()).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
}
if (lastAcked > 0 && topology.epoch() > lastAcked + 1)
{
- epochs.acknowledgeFuture(lastAcked + 1).addCallback(() -> reportTopology(topology, startSync));
+ epochs.acknowledgeFuture(lastAcked + 1).addCallback(() -> reportTopology(topology, startSync, isLoad));
return;
}
logger.trace("Epoch {} received by {}", topology.epoch(), localId);
@@ -296,13 +296,13 @@ public abstract class AbstractConfigurationService<EpochState extends AbstractCo
epochs.receive(topology);
topologyUpdatePreListenerNotify(topology);
for (Listener listener : listeners)
- listener.onTopologyUpdate(topology, startSync);
+ listener.onTopologyUpdate(topology, isLoad, startSync);
topologyUpdatePostListenerNotify(topology);
}
public synchronized void reportTopology(Topology topology)
{
- reportTopology(topology, true);
+ reportTopology(topology, false, true);
}
protected void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch) {}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index a013a503..e01614c2 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -1450,7 +1450,7 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- protected void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore)
+ protected void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore)
{
registerHistoricalTransactions(deps, (key, txnId) -> safeStore.get(key).registerHistorical(safeStore, txnId));
}
diff --git a/accord-core/src/main/java/accord/impl/MajorityDepsFetcher.java b/accord-core/src/main/java/accord/impl/MajorityDepsFetcher.java
index 0c3bd23e..dcb94801 100644
--- a/accord-core/src/main/java/accord/impl/MajorityDepsFetcher.java
+++ b/accord-core/src/main/java/accord/impl/MajorityDepsFetcher.java
@@ -18,7 +18,10 @@
package accord.impl;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -36,6 +39,7 @@ import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
+import accord.utils.async.AsyncResult;
import static accord.local.KeyHistory.COMMANDS;
import static accord.local.PreLoadContext.contextFor;
@@ -62,6 +66,7 @@ public class MajorityDepsFetcher
{
final CommandStore commandStore;
final Range range;
+ final List<AsyncResult.Settable<Void>> waiting;
long epoch;
boolean defunct;
@@ -71,10 +76,11 @@ public class MajorityDepsFetcher
Scheduler.Scheduled scheduled;
long retryDelayMicros = defaultRetryDelayMicros;
- private ShardScheduler(CommandStore commandStore, Range range, long epoch)
+ private ShardScheduler(CommandStore commandStore, Range range, List<AsyncResult.Settable<Void>> waiting, long epoch)
{
this.commandStore = commandStore;
this.range = range;
+ this.waiting = waiting;
this.numberOfSplits = targetShardSplits;
this.epoch = epoch;
}
@@ -137,7 +143,7 @@ public class MajorityDepsFetcher
nextIndex = i;
}
- Runnable schedule = () -> start(Ranges.of(range), nextIndex);
+ Runnable schedule = () -> start(range, nextIndex);
if (scheduleAt <= nowMicros) schedule.run();
else scheduled = node.scheduler().once(schedule, scheduleAt - nowMicros, MICROSECONDS);
}
@@ -147,7 +153,7 @@ public class MajorityDepsFetcher
* The first step for coordinating shard durable is to run an exclusive sync point
* the result of which can then be used to run
*/
- private void start(Ranges ranges, int nextIndex)
+ private void start(Range slice, int nextIndex)
{
TxnId id = TxnId.fromValues(epoch - 1, 0, node.id());
Timestamp before = Timestamp.minForEpoch(epoch);
@@ -156,19 +162,19 @@ public class MajorityDepsFetcher
if (withEpochFailure != null)
{
// don't wait on epoch failure - we aren't the cause of any problems
- start(ranges, nextIndex);
+ start(slice, nextIndex);
Throwable wrapped = CoordinationFailed.wrap(withEpochFailure);
logger.trace("Exception waiting for epoch before coordinating exclusive sync point for local shard durability, epoch " + id.epoch(), wrapped);
node.agent().onUncaughtException(wrapped);
return;
}
scheduled = null;
- FullRoute<Range> route = (FullRoute<Range>) node.computeRoute(id, ranges);
- logger.debug("Fetching deps to sync epoch {} for range {}", epoch, ranges);
+ FullRoute<Range> route = (FullRoute<Range>) node.computeRoute(id, Ranges.of(slice));
+ logger.debug("Fetching deps to sync epoch {} for range {}", epoch, slice);
CollectCalculatedDeps.withCalculatedDeps(node, id, route, route, before, (deps, fail) -> {
if (fail != null)
{
- logger.warn("Failed to fetch deps for syncing epoch {} for {}", epoch, ranges, fail);
+ logger.warn("Failed to fetch deps for syncing epoch {} for {}", epoch, slice, fail);
retry();
}
else
@@ -176,12 +182,12 @@ public class MajorityDepsFetcher
// TODO (correctness) : PreLoadContext only works with Seekables, which doesn't allow mixing Keys and Ranges... But Deps has both Keys AND Ranges!
// ATM all known implementations store ranges in-memory, but this will not be true soon, so this will need to be addressed
commandStore.execute(contextFor(null, deps.keyDeps.keys(), COMMANDS), safeStore -> {
- safeStore.registerHistoricalTransactions(deps);
+ safeStore.registerHistoricalTransactions(epoch, slice, deps);
}).begin((success, fail2) -> {
if (fail2 != null)
{
retry();
- logger.warn("Failed to apply deps for syncing epoch {} for range {}", epoch, ranges, fail2);
+ logger.warn("Failed to apply deps for syncing epoch {} for range {}", epoch, slice, fail2);
}
else
{
@@ -192,9 +198,10 @@ public class MajorityDepsFetcher
index = nextIndex;
if (index >= numberOfSplits)
{
- logger.info("Successfully fetched majority deps for {} at epoch {}", range, epoch);
+ waiting.forEach(w -> w.trySuccess(null));
+ logger.info("Successfully fetched majority deps for {} at epoch {}", this.range, epoch);
defunct = true;
- shardSchedulers.remove(range, ShardScheduler.this);
+ shardSchedulers.remove(this.range, ShardScheduler.this);
}
else
{
@@ -220,6 +227,16 @@ public class MajorityDepsFetcher
this.node = node;
}
+ public synchronized void cancel(Range range, long epoch)
+ {
+ ShardScheduler scheduler = shardSchedulers.get(range);
+ if (scheduler == null || scheduler.epoch > epoch)
+ return;
+
+ scheduler.markDefunct();
+ shardSchedulers.remove(range);
+ }
+
public void setTargetShardSplits(int targetShardSplits)
{
this.targetShardSplits = targetShardSplits;
@@ -235,16 +252,19 @@ public class MajorityDepsFetcher
this.maxRetryDelayMicros = units.toMicros(retryDelay);
}
- public synchronized void fetchMajorityDeps(CommandStore commandStore, Range range, long epoch)
+ public synchronized void fetchMajorityDeps(CommandStore commandStore, Range range, long epoch, AsyncResult.Settable<Void> waiting)
{
ShardScheduler scheduler = shardSchedulers.get(range);
+ List<AsyncResult.Settable<Void>> waitingList = Collections.singletonList(waiting);
if (scheduler != null)
{
if (scheduler.epoch >= epoch)
return;
scheduler.markDefunct();
+ waitingList = new ArrayList<>(waitingList);
+ waitingList.addAll(scheduler.waiting);
}
- scheduler = new ShardScheduler(commandStore, range, epoch);
+ scheduler = new ShardScheduler(commandStore, range, waitingList, epoch);
scheduler.schedule();
}
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index a50777bf..29edb281 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -35,10 +35,13 @@ import accord.utils.async.AsyncChain;
import accord.api.ConfigurationService.EpochReady;
import accord.utils.DeterministicIdentitySet;
import accord.utils.Invariants;
+import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
@@ -235,7 +238,7 @@ public abstract class CommandStore implements AgentExecutor
public abstract <T> AsyncChain<T> submit(PreLoadContext context, Function<? super SafeCommandStore, T> apply);
public abstract void shutdown();
- protected abstract void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore);
+ protected abstract void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore);
protected void unsafeSetRejectBefore(RejectBefore newRejectBefore)
{
@@ -465,22 +468,36 @@ public abstract class CommandStore implements AgentExecutor
* So, the outer future's success is sufficient for the topology to be acknowledged, and the inner future for the
* bootstrap to be complete.
*/
- protected Supplier<EpochReady> sync(Node node, Ranges ranges, long epoch)
+ protected Supplier<EpochReady> sync(Node node, Ranges ranges, long epoch, boolean isLoad)
{
- return () -> {
- AsyncResults.SettableResult<Void> whenDone = new AsyncResults.SettableResult<>();
- fetchMajorityDeps(whenDone, node, epoch, ranges);
- return new EpochReady(epoch, DONE, whenDone, whenDone, whenDone);
- };
+ return () -> syncInternal(node, ranges, epoch, isLoad);
+ }
+
+ protected EpochReady syncInternal(Node node, Ranges ranges, long epoch, boolean isLoad)
+ {
+ AsyncResults.SettableResult<Void> whenDone = new AsyncResults.SettableResult<>();
+ fetchMajorityDeps(whenDone, node, epoch, ranges);
+ return new EpochReady(epoch, DONE, whenDone, whenDone, whenDone);
}
private MajorityDepsFetcher fetcher;
+ protected void cancelFetch(Range range, long epoch)
+ {
+ if (fetcher != null)
+ fetcher.cancel(range, epoch);
+ }
// TODO (required, correctness): replace with a simple wait on suitable exclusive sync point(s)
- private void fetchMajorityDeps(AsyncResults.SettableResult<Void> coordination, Node node, long epoch, Ranges ranges)
+ private void fetchMajorityDeps(AsyncResult.Settable<Void> coordination, Node node, long epoch, Ranges ranges)
{
if (fetcher == null) fetcher = new MajorityDepsFetcher(node);
+ List<AsyncResult.Settable<Void>> waiting = new ArrayList<>();
for (Range range : ranges)
- fetcher.fetchMajorityDeps(this, range, epoch);
+ {
+ AsyncResult.Settable<Void> rangeComplete = AsyncResults.settable();
+ fetcher.fetchMajorityDeps(this, range, epoch, rangeComplete);
+ waiting.add(rangeComplete);
+ }
+ AsyncChains.reduce(waiting, (a, b) -> null).begin(coordination.settingCallback());
}
Supplier<EpochReady> unbootstrap(long epoch, Ranges removedRanges)
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
index e10d80d3..3804254c 100644
--- a/accord-core/src/main/java/accord/local/CommandStores.java
+++ b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -402,7 +402,7 @@ public abstract class CommandStores
return newLocalTopology.epoch() != 1;
}
- private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Topology newTopology, boolean startSync)
+ private synchronized TopologyUpdate updateTopology(Node node, Snapshot prev, Topology newTopology, boolean isLoad, boolean startSync)
{
checkArgument(!newTopology.isSubset(), "Use full topology for CommandStores.updateTopology");
@@ -451,7 +451,7 @@ public abstract class CommandStores
// ranges can be empty when ranges are lost or consolidated across epochs.
if (epoch > 1 && startSync && requiresSync(ranges, prev.global, newTopology))
{
- bootstrapUpdates.add(shard.store.sync(node, ranges, epoch));
+ bootstrapUpdates.add(shard.store.sync(node, ranges, epoch, isLoad));
}
result.add(shard);
}
@@ -687,9 +687,9 @@ public abstract class CommandStores
return chain == null ? AsyncChains.success(null) : chain;
}
- public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean startSync)
+ public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean isLoad, boolean startSync)
{
- TopologyUpdate update = updateTopology(node, current, newTopology, startSync);
+ TopologyUpdate update = updateTopology(node, current, newTopology, isLoad, startSync);
current = update.snapshot;
return update.bootstrap;
}
diff --git a/accord-core/src/main/java/accord/local/Node.java b/accord-core/src/main/java/accord/local/Node.java
index 23b4728f..f3f46c82 100644
--- a/accord-core/src/main/java/accord/local/Node.java
+++ b/accord-core/src/main/java/accord/local/Node.java
@@ -231,7 +231,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
@VisibleForTesting
public AsyncResult<Void> unsafeStart()
{
- EpochReady ready = onTopologyUpdateInternal(configService.currentTopology(), false);
+ EpochReady ready = onTopologyUpdateInternal(configService.currentTopology(), false, false);
ready.coordination.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
configService.acknowledgeEpoch(ready, false);
return ready.metadata;
@@ -302,9 +302,9 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return topology().epoch();
}
- private synchronized EpochReady onTopologyUpdateInternal(Topology topology, boolean startSync)
+ private synchronized EpochReady onTopologyUpdateInternal(Topology topology, boolean isLoad, boolean startSync)
{
- Supplier<EpochReady> bootstrap = commandStores.updateTopology(this, topology, startSync);
+ Supplier<EpochReady> bootstrap = commandStores.updateTopology(this, topology, isLoad, startSync);
Supplier<EpochReady> ordering = () -> {
if (this.topology.isEmpty()) return bootstrap.get();
return order(this.topology.epochReady(topology.epoch() - 1), bootstrap.get());
@@ -325,11 +325,11 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
}
@Override
- public synchronized AsyncResult<Void> onTopologyUpdate(Topology topology, boolean startSync)
+ public synchronized AsyncResult<Void> onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync)
{
if (topology.epoch() <= this.topology.epoch())
return AsyncResults.success(null);
- EpochReady ready = onTopologyUpdateInternal(topology, startSync);
+ EpochReady ready = onTopologyUpdateInternal(topology, isLoad, startSync);
ready.coordination.addCallback(() -> this.topology.onEpochSyncComplete(id, topology.epoch()));
configService.acknowledgeEpoch(ready, startSync);
return ready.coordination;
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 8ea32a74..5c80701b 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -32,6 +32,7 @@ import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.AbstractUnseekableKeys;
import accord.primitives.Deps;
import accord.primitives.Participants;
+import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.RoutingKeys;
@@ -264,9 +265,9 @@ public abstract class SafeCommandStore
commandStore().unsafeSetRangesForEpoch(rangesForEpoch);
}
- public void registerHistoricalTransactions(Deps deps)
+ public void registerHistoricalTransactions(long epoch, Range range, Deps deps)
{
- commandStore().registerHistoricalTransactions(deps, this);
+ commandStore().registerHistoricalTransactions(range, deps, this);
}
public void updateCommandsForKey(Command prev, Command next)
diff --git a/accord-core/src/test/java/accord/impl/AbstractConfigurationServiceTest.java b/accord-core/src/test/java/accord/impl/AbstractConfigurationServiceTest.java
index 59eb56b4..4a34a4d4 100644
--- a/accord-core/src/test/java/accord/impl/AbstractConfigurationServiceTest.java
+++ b/accord-core/src/test/java/accord/impl/AbstractConfigurationServiceTest.java
@@ -61,7 +61,7 @@ public class AbstractConfigurationServiceTest
}
@Override
- public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean startSync)
+ public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync)
{
if (topologies.put(topology.epoch(), topology) != null)
Assertions.fail("Received topology twice for epoch " + topology.epoch());
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index ebad71ac..80d26851 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -51,6 +51,7 @@ import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
+import accord.primitives.Range;
import accord.primitives.SaveStatus;
import accord.primitives.Status.Durability;
import accord.local.cfk.SafeCommandsForKey;
@@ -398,7 +399,7 @@ public class RemoteListenersTest
@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) { return null; }
@Override public void shutdown() {}
- @Override protected void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore) {}
+ @Override protected void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore) {}
@Override public <T> AsyncChain<T> submit(Callable<T> task) { return null; }
}
@@ -433,6 +434,6 @@ public class RemoteListenersTest
@Override public ProgressLog progressLog() { return null; }
@Override public NodeCommandStoreService node() { return null; }
@Override public CommandStores.RangesForEpoch ranges() { return null; }
- @Override public void registerHistoricalTransactions(Deps deps) { }
+ @Override public void registerHistoricalTransactions(long epoch, Range range, Deps deps) { }
}
}
diff --git a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
index b3f9370e..b5747eb8 100644
--- a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
+++ b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
@@ -288,10 +288,10 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
}
@Override
- protected void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore)
+ protected void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore)
{
journal.registerHistoricalTransactions(id(), deps);
- super.registerHistoricalTransactions(deps, safeStore);
+ super.registerHistoricalTransactions(range, deps, safeStore);
}
@Override
diff --git a/accord-core/src/test/java/accord/impl/mock/MockCluster.java b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
index 93743b21..b3da2479 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockCluster.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
@@ -145,7 +145,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
DurableBefore.NOOP_PERSISTER,
localConfig);
awaitUninterruptibly(node.unsafeStart());
- node.onTopologyUpdate(topology, true);
+ node.onTopologyUpdate(topology, false, true);
return node;
}
diff --git a/accord-core/src/test/java/accord/impl/mock/MockConfigurationService.java b/accord-core/src/test/java/accord/impl/mock/MockConfigurationService.java
index fbcd107d..a892bb8a 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockConfigurationService.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockConfigurationService.java
@@ -115,7 +115,7 @@ public class MockConfigurationService implements TestableConfigurationService
List<AsyncResult<Void>> futures = new ArrayList<>();
for (Listener listener : listeners)
- futures.add(listener.onTopologyUpdate(topology, true));
+ futures.add(listener.onTopologyUpdate(topology, false, true));
AsyncResult<Void> result = futures.isEmpty()
? AsyncResults.success(null)
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
index 37c2e243..57d508e8 100644
--- a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
+++ b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -110,7 +110,7 @@ public class ImmutableCommandTest
DurableBefore.NOOP_PERSISTER,
localConfig);
awaitUninterruptibly(node.unsafeStart());
- node.onTopologyUpdate(storeSupport.local.get(), true);
+ node.onTopologyUpdate(storeSupport.local.get(), false, true);
return node;
}
diff --git a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
index d8bb7bfa..710780fe 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -961,7 +961,7 @@ public class CommandsForKeyTest
}
@Override
- protected void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore)
+ protected void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore)
{
throw new UnsupportedOperationException();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org