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

CEP-15: (Accord) When nodes are removed from a cluster, need to update topology tracking to avoid being blocked 

patch by David Capwell; reviewed by Blake Eggleston for CASSANDRA-19719
This commit is contained in:
David Capwell 2024-06-18 13:55:38 -07:00
parent ac58a62c51
commit 0d4c0961be
23 changed files with 1230 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
build.xml
View File

@ -1289,13 +1289,12 @@
<condition property="maxMemory" value="8G" else="4G">
<equals arg1="${test.classlistprefix}" arg2="distributed"/>
</condition>
<testmacrohelper inputdir="${test.dir}/${test.classlistprefix}" filelist="@{test.file.list}" exclude="**/*.java" timeout="${test.timeout}">
<testmacrohelper inputdir="${test.dir}/${test.classlistprefix}" filelist="@{test.file.list}" exclude="**/*.java" timeout="${test.timeout}" maxmemory="${maxMemory}">
<jvmarg value="-Dlegacy-sstable-root=${test.data}/legacy-sstables"/>
<jvmarg value="-Dinvalid-legacy-sstable-root=${test.data}/invalid-legacy-sstables"/>
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
<jvmarg value="-Xmx${maxMemory}"/>
</testmacrohelper>
</sequential>
</macrodef>

2
modules/accord

@ -1 +1 @@
Subproject commit f1f5ea5ccbd6e0a8abf579a4331fa84a1b3d9f95
Subproject commit 694ae39e2e00075bdabd47632dced0db12a9981d

2
src/java/org/apache/cassandra/repair/RepairCoordinator.java
View File

@ -129,7 +129,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
{
this.ctx = ctx;
this.validationScheduler = Scheduler.build(DatabaseDescriptor.getConcurrentMerkleTreeRequests());
this.state = new CoordinatorState(ctx.clock(), cmd, keyspace, options);
this.state = new CoordinatorState(ctx, cmd, keyspace, options);
this.tag = "repair:" + cmd;
this.validColumnFamilies = validColumnFamilies;
this.getLocalReplicas = getLocalReplicas;

2
src/java/org/apache/cassandra/repair/RepairSession.java
View File

@ -182,7 +182,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
this.paxosOnly = paxosOnly;
this.isConsensusMigration = isConsensusMigration;
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";
this.state = new SessionState(ctx.clock(), parentRepairSession, keyspace, cfnames, commonRange);
this.state = new SessionState(ctx, parentRepairSession, keyspace, cfnames, commonRange);
this.parallelismDegree = parallelismDegree;
this.isIncremental = isIncremental;
this.previewKind = previewKind;

5
src/java/org/apache/cassandra/repair/SharedContext.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.TimeUUID;
/**
* Access methods to shared resources and services.
@ -81,6 +82,10 @@ public interface SharedContext
TableRepairManager repairManager(ColumnFamilyStore store);
StreamExecutor streamExecutor();
PaxosRepairState paxosRepairState();
default Supplier<TimeUUID> timeUUID()
{
return TimeUUID.Generator::nextTimeUUID;
}
class Global implements SharedContext
{

8
src/java/org/apache/cassandra/repair/state/CoordinatorState.java
View File

@ -30,12 +30,10 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.CommonRange;
import org.apache.cassandra.repair.RepairCoordinator;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
public class CoordinatorState extends AbstractState<CoordinatorState.State, TimeUUID>
{
public enum State
@ -56,9 +54,9 @@ public class CoordinatorState extends AbstractState<CoordinatorState.State, Time
// API to split function calls for phase changes from getting the state
public final Phase phase = new Phase();
public CoordinatorState(Clock clock, int cmd, String keyspace, RepairOption options)
public CoordinatorState(SharedContext ctx, int cmd, String keyspace, RepairOption options)
{
super(clock, nextTimeUUID(), State.class);
super(ctx.clock(), ctx.timeUUID().get(), State.class);
this.cmd = cmd;
this.keyspace = Objects.requireNonNull(keyspace);
this.options = Objects.requireNonNull(options);

8
src/java/org/apache/cassandra/repair/state/SessionState.java
View File

@ -26,11 +26,9 @@ import java.util.stream.Collectors;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.CommonRange;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
public class SessionState extends AbstractState<SessionState.State, TimeUUID>
{
public enum State
@ -46,9 +44,9 @@ public class SessionState extends AbstractState<SessionState.State, TimeUUID>
public final Phase phase = new Phase();
public SessionState(Clock clock, TimeUUID parentRepairSession, String keyspace, String[] cfnames, CommonRange commonRange)
public SessionState(SharedContext ctx, TimeUUID parentRepairSession, String keyspace, String[] cfnames, CommonRange commonRange)
{
super(clock, nextTimeUUID(), State.class);
super(ctx.clock(), ctx.timeUUID().get(), State.class);
this.parentRepairSession = parentRepairSession;
this.keyspace = keyspace;
this.cfnames = cfnames;

138
src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
View File

@ -34,6 +34,8 @@ import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.agrona.collections.LongArrayList;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.concurrent.Stage;
@ -57,6 +59,7 @@ import static org.apache.cassandra.utils.Simulate.With.MONITORS;
public class AccordConfigurationService extends AbstractConfigurationService<AccordConfigurationService.EpochState, AccordConfigurationService.EpochHistory> implements ChangeListener, AccordEndpointMapper, AccordSyncPropagator.Listener, Shutdownable
{
private final AccordSyncPropagator syncPropagator;
private final DiskStateManager diskStateManager;
private EpochDiskState diskState = EpochDiskState.EMPTY;
@ -114,15 +117,88 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
}
}
public AccordConfigurationService(Node.Id node, MessageDelivery messagingService, IFailureDetector failureDetector)
@VisibleForTesting
interface DiskStateManager
{
EpochDiskState loadTopologies(AccordKeyspace.TopologyLoadConsumer consumer);
EpochDiskState setNotifyingLocalSync(long epoch, Set<Node.Id> pending, EpochDiskState diskState);
EpochDiskState setCompletedLocalSync(long epoch, EpochDiskState diskState);
EpochDiskState markLocalSyncAck(Node.Id id, long epoch, EpochDiskState diskState);
EpochDiskState saveTopology(Topology topology, EpochDiskState diskState);
EpochDiskState markRemoteTopologySync(Node.Id node, long epoch, EpochDiskState diskState);
EpochDiskState markClosed(Ranges ranges, long epoch, EpochDiskState diskState);
EpochDiskState truncateTopologyUntil(long epoch, EpochDiskState diskState);
}
enum SystemTableDiskStateManager implements DiskStateManager
{
instance;
@Override
public EpochDiskState loadTopologies(AccordKeyspace.TopologyLoadConsumer consumer)
{
return AccordKeyspace.loadTopologies(consumer);
}
@Override
public EpochDiskState setNotifyingLocalSync(long epoch, Set<Node.Id> notify, EpochDiskState diskState)
{
return AccordKeyspace.setNotifyingLocalSync(epoch, notify, diskState);
}
@Override
public EpochDiskState setCompletedLocalSync(long epoch, EpochDiskState diskState)
{
return AccordKeyspace.setCompletedLocalSync(epoch, diskState);
}
@Override
public EpochDiskState markLocalSyncAck(Node.Id id, long epoch, EpochDiskState diskState)
{
return AccordKeyspace.markLocalSyncAck(id, epoch, diskState);
}
@Override
public EpochDiskState saveTopology(Topology topology, EpochDiskState diskState)
{
return AccordKeyspace.saveTopology(topology, diskState);
}
@Override
public EpochDiskState markRemoteTopologySync(Node.Id node, long epoch, EpochDiskState diskState)
{
return AccordKeyspace.markRemoteTopologySync(node, epoch, diskState);
}
@Override
public EpochDiskState markClosed(Ranges ranges, long epoch, EpochDiskState diskState)
{
return AccordKeyspace.markClosed(ranges, epoch, diskState);
}
@Override
public EpochDiskState truncateTopologyUntil(long epoch, EpochDiskState diskState)
{
return AccordKeyspace.truncateTopologyUntil(epoch, diskState);
}
}
public AccordConfigurationService(Node.Id node, MessageDelivery messagingService, IFailureDetector failureDetector, DiskStateManager diskStateManager, ScheduledExecutorPlus scheduledTasks)
{
super(node);
this.syncPropagator = new AccordSyncPropagator(localId, this, messagingService, failureDetector, ScheduledExecutors.scheduledTasks, this);
this.syncPropagator = new AccordSyncPropagator(localId, this, messagingService, failureDetector, scheduledTasks, this);
this.diskStateManager = diskStateManager;
}
public AccordConfigurationService(Node.Id node)
{
this(node, MessagingService.instance(), FailureDetector.instance);
this(node, MessagingService.instance(), FailureDetector.instance, SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
}
@Override
@ -137,7 +213,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
state = State.LOADING;
updateMapping(ClusterMetadata.current());
EndpointMapping snapshot = mapping;
diskState = AccordKeyspace.loadTopologies(((epoch, topology, syncStatus, pendingSyncNotify, remoteSyncComplete, closed, redundant) -> {
diskState = diskStateManager.loadTopologies(((epoch, topology, syncStatus, pendingSyncNotify, remoteSyncComplete, closed, redundant) -> {
if (topology != null)
reportTopology(topology, syncStatus == SyncStatus.NOT_STARTED);
@ -221,12 +297,41 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
synchronized (AccordConfigurationService.this)
{
updateMapping(metadata);
reportTopology(AccordTopology.createAccordTopology(metadata));
Topology topology = AccordTopology.createAccordTopology(metadata);
Topology current = isEmpty() ? Topology.EMPTY : currentTopology();
reportTopology(topology);
Sets.SetView<Node.Id> removedNodes = Sets.difference(current.nodes(), topology.nodes());
if (!removedNodes.isEmpty())
onNodesRemoved(topology.epoch(), removedNodes);
}
});
}
private void maybeReportMetadata(ClusterMetadata metadata)
private synchronized void onNodesRemoved(long epoch, Set<Node.Id> removed)
{
syncPropagator.onNodesRemoved(removed);
for (long oldEpoch : nonCompletedEpochsBefore(epoch))
{
for (Node.Id node : removed)
receiveRemoteSyncComplete(node, oldEpoch);
}
listeners.forEach(l -> l.onRemoveNodes(epoch, removed));
}
private long[] nonCompletedEpochsBefore(long max)
{
LongArrayList notComplete = new LongArrayList();
for (long epoch = epochs.minEpoch(); epoch <= max && epoch <= epochs.maxEpoch(); epoch++)
{
EpochSnapshot snapshot = getEpochSnapshot(epoch);
if (snapshot.syncStatus != SyncStatus.COMPLETED)
notComplete.add(epoch);
}
return notComplete.toLongArray();
}
@VisibleForTesting
void maybeReportMetadata(ClusterMetadata metadata)
{
// don't report metadata until the previous one has been acknowledged
synchronized (this)
@ -265,7 +370,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
return;
Set<Node.Id> notify = topology.nodes().stream().filter(i -> !localId.equals(i)).collect(Collectors.toSet());
diskState = AccordKeyspace.setNotifyingLocalSync(epoch, notify, diskState);
diskState = diskStateManager.setNotifyingLocalSync(epoch, notify, diskState);
epochState.setSyncStatus(SyncStatus.NOTIFYING);
syncPropagator.reportSyncComplete(epoch, notify, localId);
}
@ -276,7 +381,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
EpochState epochState = getOrCreateEpochState(epoch);
if (epochState.syncStatus != SyncStatus.NOTIFYING)
return;
diskState = AccordKeyspace.markLocalSyncAck(id, epoch, diskState);
diskState = diskStateManager.markLocalSyncAck(id, epoch, diskState);
}
@Override
@ -284,21 +389,21 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
{
EpochState epochState = getOrCreateEpochState(epoch);
epochState.setSyncStatus(SyncStatus.COMPLETED);
diskState = AccordKeyspace.setCompletedLocalSync(epoch, diskState);
diskState = diskStateManager.setCompletedLocalSync(epoch, diskState);
}
@Override
protected synchronized void topologyUpdatePreListenerNotify(Topology topology)
{
if (state == State.STARTED)
diskState = AccordKeyspace.saveTopology(topology, diskState);
diskState = diskStateManager.saveTopology(topology, diskState);
}
@Override
protected synchronized void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch)
{
if (state == State.STARTED)
diskState = AccordKeyspace.markRemoteTopologySync(node, epoch, diskState);
diskState = diskStateManager.markRemoteTopologySync(node, epoch, diskState);
}
@Override
@ -309,6 +414,11 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
syncPropagator.reportClosed(epoch, topology.nodes(), ranges);
}
public AccordSyncPropagator syncPropagator()
{
return syncPropagator;
}
@Override
public synchronized void reportEpochRedundant(Ranges ranges, long epoch)
{
@ -321,14 +431,14 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
@Override
public synchronized void receiveClosed(Ranges ranges, long epoch)
{
diskState = AccordKeyspace.markClosed(ranges, epoch, diskState);
diskState = diskStateManager.markClosed(ranges, epoch, diskState);
super.receiveClosed(ranges, epoch);
}
@Override
public synchronized void receiveRedundant(Ranges ranges, long epoch)
{
diskState = AccordKeyspace.markClosed(ranges, epoch, diskState);
diskState = diskStateManager.markClosed(ranges, epoch, diskState);
super.receiveRedundant(ranges, epoch);
}
@ -342,7 +452,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
protected synchronized void truncateTopologiesPostListenerNotify(long epoch)
{
if (state == State.STARTED)
diskState = AccordKeyspace.truncateTopologyUntil(epoch, diskState);
diskState = diskStateManager.truncateTopologyUntil(epoch, diskState);
}
private void checkStarted()

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

@ -1587,7 +1587,8 @@ public class AccordKeyspace
return minEpoch == maxEpoch && maxEpoch == 0;
}
private EpochDiskState withNewMaxEpoch(long epoch)
@VisibleForTesting
EpochDiskState withNewMaxEpoch(long epoch)
{
Invariants.checkArgument(epoch > maxEpoch, "Epoch %d <= %d (max)", epoch, maxEpoch);
return EpochDiskState.create(Math.max(1, minEpoch), epoch);

82
src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@ -29,6 +30,9 @@ import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.Node;
import accord.messages.SimpleReply;
import accord.primitives.Ranges;
@ -38,6 +42,7 @@ import org.agrona.collections.Long2ObjectHashMap;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@ -50,6 +55,7 @@ import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.CollectionSerializers;
import static org.apache.cassandra.utils.CollectionSerializers.newListSerializer;
@ -59,6 +65,8 @@ import static org.apache.cassandra.utils.CollectionSerializers.newListSerializer
*/
public class AccordSyncPropagator
{
private static final Logger logger = LoggerFactory.getLogger(AccordSyncPropagator.class);
public static final IVerbHandler<List<Notification>> verbHandler = message -> {
if (!AccordService.isSetup())
return;
@ -120,6 +128,11 @@ public class AccordSyncPropagator
return new Notification(epoch, Collections.emptySet(), Ranges.EMPTY, addRedundant);
}
boolean isEmpty()
{
return syncComplete.isEmpty() && closed.isEmpty() && redundant.isEmpty();
}
boolean ack(Notification notification)
{
if (!notification.syncComplete.isEmpty())
@ -201,6 +214,15 @@ public class AccordSyncPropagator
return !pending.isEmpty();
}
synchronized boolean hasPending(long epoch)
{
if (pending.isEmpty()) return false;
return pending.values().stream().allMatch(n -> {
PendingEpoch p = n.get(epoch);
return p != null && !p.isEmpty();
});
}
@Override
public String toString()
{
@ -210,6 +232,28 @@ public class AccordSyncPropagator
'}';
}
public synchronized void onNodesRemoved(Set<Node.Id> removed)
{
for (Node.Id node : removed)
{
PendingEpochs pendingEpochs = pending.get(node.id);
if (pendingEpochs == null) continue;
long[] toComplete = new long[pendingEpochs.size()];
Long2ObjectHashMap<PendingEpoch>.KeyIterator it = pendingEpochs.keySet().iterator();
for (int i = 0; it.hasNext(); i++)
toComplete[i] = it.nextLong();
Arrays.sort(toComplete);
for (long epoch : toComplete)
listener.onEndpointAck(node, epoch);
pending.remove(node.id);
for (long epoch : toComplete)
{
if (hasSyncCompletedFor(epoch))
listener.onComplete(epoch);
}
}
}
public void reportSyncComplete(long epoch, Collection<Node.Id> notify, Node.Id syncCompleteId)
{
if (notify.isEmpty())
@ -258,17 +302,13 @@ public class AccordSyncPropagator
private boolean notify(Node.Id to, List<Notification> notifications)
{
InetAddressAndPort toEp = endpointMapper.mappedEndpoint(to);
if (!failureDetector.isAlive(toEp))
{
scheduler.schedule(() -> notify(to, notifications), 1, TimeUnit.MINUTES);
return false;
}
Message<List<Notification>> msg = Message.out(Verb.ACCORD_SYNC_NOTIFY_REQ, notifications);
messagingService.sendWithCallback(msg, toEp, new RequestCallback<SimpleReply>(){
RequestCallback<SimpleReply> cb = new RequestCallback<>()
{
@Override
public void onResponse(Message<SimpleReply> msg)
{
Invariants.checkState(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg);
Invariants.checkState(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg);
Set<Long> completedEpochs = new HashSet<>();
// TODO review is it a good idea to call the listener while not holding the `AccordSyncPropagator` lock?
synchronized (AccordSyncPropagator.this)
@ -304,7 +344,22 @@ public class AccordSyncPropagator
{
return true;
}
});
};
if (!failureDetector.isAlive(toEp))
{
// was the endpoint removed from membership?
ClusterMetadata metadata = ClusterMetadata.current();
if (Gossiper.instance.getEndpointStateForEndpoint(toEp) == null && !metadata.directory.allJoinedEndpoints().contains(toEp) && !metadata.fullCMSMembers().contains(toEp))
{
// endpoint no longer exists...
cb.onResponse(msg.responseWith(SimpleReply.Ok));
return true;
}
logger.warn("Node{} is not alive, unable to notify of {}", to, notifications);
scheduler.schedule(() -> notify(to, notifications), 1, TimeUnit.MINUTES);
return false;
}
messagingService.sendWithCallback(msg, toEp, cb);
return true;
}
@ -352,5 +407,16 @@ public class AccordSyncPropagator
this.closed = closed;
this.redundant = redundant;
}
@Override
public String toString()
{
return "Notification{" +
"epoch=" + epoch +
", syncComplete=" + syncComplete +
", closed=" + closed +
", redundant=" + redundant +
'}';
}
}
}

2
test/unit/org/apache/cassandra/concurrent/SimulatedExecutorFactory.java
View File

@ -101,7 +101,7 @@ public class SimulatedExecutorFactory implements ExecutorFactory, Clock
public SimulatedExecutorFactory(RandomSource rs, Consumer<Throwable> onError)
{
this(rs, toGen(Generators.TIMESTAMP_GEN.map(Timestamp::getTime)).mapToLong(TimeUnit.MILLISECONDS::toNanos).next(rs), onError);
this(rs, toGen(Generators.TIMESTAMP_GEN.map(Timestamp::getTime)).mapToLong(TimeUnit.MILLISECONDS::toNanos).nextLong(rs), onError);
}
public SimulatedExecutorFactory(RandomSource rs)

5
test/unit/org/apache/cassandra/db/virtual/LocalRepairTablesTest.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.CommonRange;
import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.repair.RepairCoordinator;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.repair.messages.PrepareMessage;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.state.Completable;
@ -291,7 +292,7 @@ public class LocalRepairTablesTest extends CQLTester
private static CoordinatorState coordinator()
{
RepairOption options = RepairOption.parse(Collections.emptyMap(), DatabaseDescriptor.getPartitioner());
CoordinatorState state = new CoordinatorState(Clock.Global.clock(), 0, "test", options);
CoordinatorState state = new CoordinatorState(SharedContext.Global.instance, 0, "test", options);
ActiveRepairService.instance().register(state);
return state;
}
@ -299,7 +300,7 @@ public class LocalRepairTablesTest extends CQLTester
private static SessionState session()
{
CoordinatorState parent = coordinator();
SessionState state = new SessionState(Clock.Global.clock(), parent.id, REPAIR_KS, new String[]{ REPAIR_TABLE }, COMMON_RANGE);
SessionState state = new SessionState(SharedContext.Global.instance, parent.id, REPAIR_KS, new String[]{ REPAIR_TABLE }, COMMON_RANGE);
parent.register(state);
return state;
}

8
test/unit/org/apache/cassandra/index/accord/AccordIndexStressTest.java
View File

@ -257,11 +257,11 @@ public class AccordIndexStressTest extends CQLTester
{
case Key:
{
store = rs.pick(storeToTableToRoutingKeysToTxns.keySet());
store = rs.pickUnorderedSet(storeToTableToRoutingKeysToTxns.keySet());
var actual = this.storeToTableToRoutingKeysToTxns.get(store);
var tableToTokens = store2Table2Tokens.get(store);
table = rs.pick(actual.keySet());
table = rs.pickUnorderedSet(actual.keySet());
var tokens = tableToTokens.get(table);
var offset = rs.nextInt(0, tokens.length);
@ -274,11 +274,11 @@ public class AccordIndexStressTest extends CQLTester
break;
case Range:
{
store = rs.pick(storeToTableToRangesToTxns.keySet());
store = rs.pickUnorderedSet(storeToTableToRangesToTxns.keySet());
var tableToRangesToTxns = storeToTableToRangesToTxns.get(store);
var tableToRanges = store2Table2Ranges.get(store);
table = rs.pick(tableToRangesToTxns.keySet());
table = rs.pickUnorderedSet(tableToRangesToTxns.keySet());
var wrapper = tableToRanges.get(table);
var ranges = wrapper.ranges;
var tree = wrapper.tree;

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

@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -125,7 +126,7 @@ public class RouteIndexTest extends CQLTester.InMemory
@Override
public Gen<Command<State, ColumnFamilyStore, ?>> commands(State state)
{
Map<Gen<Command<State, ColumnFamilyStore, ?>>, Integer> possible = new HashMap<>();
Map<Gen<Command<State, ColumnFamilyStore, ?>>, Integer> possible = new LinkedHashMap<>();
possible.put(ignore -> FLUSH, 1);
possible.put(ignore -> COMPACT, 1);
possible.put(rs -> {
@ -139,9 +140,9 @@ public class RouteIndexTest extends CQLTester.InMemory
if (!state.storeToTableToRangesToTxns.isEmpty())
{
possible.put(rs -> {
int storeId = rs.pick(state.storeToTableToRangesToTxns.keySet());
int storeId = rs.pickUnorderedSet(state.storeToTableToRangesToTxns.keySet());
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pick(tables.keySet());
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
TokenRange range;
@ -154,14 +155,14 @@ public class RouteIndexTest extends CQLTester.InMemory
switch (rs.nextInt(0, 2))
{
case 0: // perfect match
range = rs.pick(distinctRanges);
range = rs.pickOrderedSet(distinctRanges);
break;
case 1: // mutli-match
{
TokenRange a = rs.pick(distinctRanges);
TokenRange b = rs.pick(distinctRanges);
TokenRange a = rs.pickOrderedSet(distinctRanges);
TokenRange b = rs.pickOrderedSet(distinctRanges);
while (a.equals(b))
b = rs.pick(distinctRanges);
b = rs.pickOrderedSet(distinctRanges);
if (b.compareTo(a) < 0)
{
TokenRange tmp = a;

1
test/unit/org/apache/cassandra/net/SimulatedMessageDelivery.java
View File

@ -31,6 +31,7 @@ import javax.annotation.Nullable;
import accord.utilsfork.Gens;
import accord.utilsfork.RandomSource;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.concurrent.AsyncPromise;

308
test/unit/org/apache/cassandra/repair/FuzzTestBase.java
View File

@ -22,7 +22,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@ -44,7 +44,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
import javax.annotation.Nullable;
@ -85,7 +84,6 @@ import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.HeartBeatState;
@ -99,12 +97,12 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.Locator;
import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.net.ConnectionType;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.net.SimulatedMessageDelivery;
import org.apache.cassandra.net.SimulatedMessageDelivery.SimulatedMessageReceiver;
import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.messages.ValidationResponse;
@ -151,8 +149,6 @@ import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.progress.ProgressEventType;
import org.assertj.core.api.Assertions;
@ -166,10 +162,10 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE
public abstract class FuzzTestBase extends CQLTester.InMemory
{
private static final int MISMATCH_NUM_PARTITIONS = 1;
private static final Gen<String> IDENTIFIER_GEN = fromQT(Generators.IDENTIFIER_GEN);
private static final Gen<String> KEYSPACE_NAME_GEN = fromQT(CassandraGenerators.KEYSPACE_NAME_GEN);
private static final Gen<TableId> TABLE_ID_GEN = fromQT(CassandraGenerators.TABLE_ID_GEN);
private static final Gen<InetAddressAndPort> ADDRESS_W_PORT = fromQT(CassandraGenerators.INET_ADDRESS_AND_PORT_GEN);
private static final Gen<String> IDENTIFIER_GEN = Generators.toGen(Generators.IDENTIFIER_GEN);
private static final Gen<String> KEYSPACE_NAME_GEN = Generators.toGen(CassandraGenerators.KEYSPACE_NAME_GEN);
private static final Gen<TableId> TABLE_ID_GEN = Generators.toGen(CassandraGenerators.TABLE_ID_GEN);
private static final Gen<InetAddressAndPort> ADDRESS_W_PORT = Generators.toGen(CassandraGenerators.INET_ADDRESS_AND_PORT_GEN);
private static boolean SETUP_SCHEMA = false;
static String KEYSPACE;
@ -494,7 +490,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
Set<Token> allTokens = new HashSet<>();
for (Range<Token> range : validator.desc.ranges)
{
Gen<Token> gen = fromQT(CassandraGenerators.tokensInRange(range));
Gen<Token> gen = Generators.toGen(CassandraGenerators.tokensInRange(range));
Set<Token> tokens = new LinkedHashSet<>();
for (int i = 0, size = rs.nextInt(1, 10); i < size; i++)
{
@ -691,15 +687,13 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
private final List<MessageListener> listeners = new ArrayList<>();
private final RandomSource rs;
private BiFunction<Node, Message<?>, Set<Faults>> allowedMessageFaults = (a, b) -> Collections.emptySet();
private final Map<Connection, LongSupplier> networkLatencies = new HashMap<>();
private final Map<Connection, Supplier<Boolean>> networkDrops = new HashMap<>();
Cluster(RandomSource rs)
{
ClockAccess.includeThreadAsOwner();
this.rs = rs;
globalExecutor = new SimulatedExecutorFactory(rs, fromQT(Generators.TIMESTAMP_GEN.map(Timestamp::getTime)).mapToLong(TimeUnit.MILLISECONDS::toNanos).next(rs));
globalExecutor = new SimulatedExecutorFactory(rs);
orderedExecutor = globalExecutor.configureSequential("ignore").build();
unorderedScheduled = globalExecutor.scheduled("ignored");
@ -720,8 +714,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
int numNodes = rs.nextInt(3, 10);
List<String> dcs = Gens.lists(IDENTIFIER_GEN).unique().ofSizeBetween(1, Math.min(10, numNodes)).next(rs);
Map<InetAddressAndPort, Node> nodes = Maps.newHashMapWithExpectedSize(numNodes);
Gen<Token> tokenGen = fromQT(CassandraGenerators.token(DatabaseDescriptor.getPartitioner()));
Gen<UUID> hostIdGen = fromQT(Generators.UUID_RANDOM_GEN);
Gen<Token> tokenGen = Generators.toGen(CassandraGenerators.token(DatabaseDescriptor.getPartitioner()));
Gen<UUID> hostIdGen = Generators.toGen(Generators.UUID_RANDOM_GEN);
Set<Token> tokens = new HashSet<>();
Set<UUID> hostIds = new HashSet<>();
for (int i = 0; i < numNodes; i++)
@ -813,214 +807,43 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
}
}
private class CallbackContext
private SimulatedMessageDelivery.Action action(InetAddressAndPort self, Message<?> msg, InetAddressAndPort to)
{
final RequestCallback callback;
private CallbackContext(RequestCallback callback)
{
this.callback = Objects.requireNonNull(callback);
}
public void onResponse(Message msg)
{
callback.onResponse(msg);
}
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
if (callback.invokeOnFailure()) callback.onFailure(from, failure);
}
boolean toSelf = self.equals(to);
Node node = nodes.get(to);
Set<Faults> allowedFaults = allowedMessageFaults.apply(node, msg);
if (allowedFaults.contains(Faults.DROP) && !toSelf && networkDrops(self, to)) return SimulatedMessageDelivery.Action.DROP_PARTITIONED;
return SimulatedMessageDelivery.Action.DELIVER;
}
private static class CallbackKey
private boolean networkDrops(InetAddressAndPort self, InetAddressAndPort to)
{
private final long id;
private final InetAddressAndPort peer;
private CallbackKey(long id, InetAddressAndPort peer)
{
this.id = id;
this.peer = peer;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
CallbackKey that = (CallbackKey) o;
return id == that.id && peer.equals(that.peer);
}
@Override
public int hashCode()
{
return Objects.hash(id, peer);
}
@Override
public String toString()
{
return "CallbackKey{" +
"id=" + id +
", peer=" + peer +
'}';
}
return networkDrops.computeIfAbsent(new Connection(self, to), ignore -> Gens.bools().biasedRepeatingRuns(rs.nextInt(1, 11) / 100.0D, rs.nextInt(3, 15)).asSupplier(rs)).get();
}
private class Messaging implements MessageDelivery
private class Messaging extends SimulatedMessageDelivery
{
final InetAddressAndPort broadcastAddressAndPort;
final Map<CallbackKey, CallbackContext> callbacks = new HashMap<>();
private Messaging(InetAddressAndPort broadcastAddressAndPort)
{
this.broadcastAddressAndPort = broadcastAddressAndPort;
}
@Override
public <REQ> void send(Message<REQ> message, InetAddressAndPort to)
{
message = message.withFrom(broadcastAddressAndPort);
maybeEnqueue(message, to, null);
}
@Override
public <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb)
{
message = message.withFrom(broadcastAddressAndPort);
maybeEnqueue(message, to, cb);
}
@Override
public <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb, ConnectionType specifyConnection)
{
message = message.withFrom(broadcastAddressAndPort);
maybeEnqueue(message, to, cb);
}
private <REQ, RSP> void maybeEnqueue(Message<REQ> message, InetAddressAndPort to, @Nullable RequestCallback<RSP> callback)
{
CallbackContext cb;
if (callback != null)
{
CallbackKey key = new CallbackKey(message.id(), to);
if (callbacks.containsKey(key))
throw new AssertionError("Message id " + message.id() + " to " + to + " already has a callback");
cb = new CallbackContext(callback);
callbacks.put(key, cb);
}
else
{
cb = null;
}
boolean toSelf = this.broadcastAddressAndPort.equals(to);
Node node = nodes.get(to);
Set<Faults> allowedFaults = allowedMessageFaults.apply(node, message);
if (allowedFaults.isEmpty())
{
// enqueue so stack overflow doesn't happen with the inlining
unorderedScheduled.submit(() -> node.handle(message));
}
else
{
Runnable enqueue = () -> {
if (!allowedFaults.contains(Faults.DELAY))
{
unorderedScheduled.submit(() -> node.handle(message));
}
else
{
if (toSelf) unorderedScheduled.submit(() -> node.handle(message));
else
unorderedScheduled.schedule(() -> node.handle(message), networkJitterNanos(to), TimeUnit.NANOSECONDS);
}
};
if (!allowedFaults.contains(Faults.DROP)) enqueue.run();
else
{
if (!toSelf && networkDrops(to))
{
// logger.warn("Dropped message {}", message);
// drop
}
else
{
enqueue.run();
}
}
if (cb != null)
{
unorderedScheduled.schedule(() -> {
CallbackContext ctx = callbacks.remove(new CallbackKey(message.id(), to));
if (ctx != null)
{
assert ctx == cb;
try
{
ctx.onFailure(to, RequestFailure.TIMEOUT);
}
catch (Throwable t)
{
failures.add(t);
}
}
}, message.verb().expiresAfterNanos(), TimeUnit.NANOSECONDS);
}
}
}
private long networkJitterNanos(InetAddressAndPort to)
{
return networkLatencies.computeIfAbsent(new Connection(broadcastAddressAndPort, to), ignore -> {
long min = TimeUnit.MICROSECONDS.toNanos(500);
long maxSmall = TimeUnit.MILLISECONDS.toNanos(5);
long max = TimeUnit.SECONDS.toNanos(5);
LongSupplier small = () -> rs.nextLong(min, maxSmall);
LongSupplier large = () -> rs.nextLong(maxSmall, max);
return Gens.bools().biasedRepeatingRuns(rs.nextInt(1, 11) / 100.0D, rs.nextInt(3, 15)).mapToLong(b -> b ? large.getAsLong() : small.getAsLong()).asLongSupplier(rs);
}).getAsLong();
}
private boolean networkDrops(InetAddressAndPort to)
{
return networkDrops.computeIfAbsent(new Connection(broadcastAddressAndPort, to), ignore -> Gens.bools().biasedRepeatingRuns(rs.nextInt(1, 11) / 100.0D, rs.nextInt(3, 15)).asSupplier(rs)).get();
}
@Override
public <REQ, RSP> Future<Message<RSP>> sendWithResult(Message<REQ> message, InetAddressAndPort to)
{
AsyncPromise<Message<RSP>> promise = new AsyncPromise<>();
sendWithCallback(message, to, new RequestCallback<RSP>()
{
@Override
public void onResponse(Message<RSP> msg)
{
promise.trySuccess(msg);
}
@Override
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
promise.tryFailure(new MessagingService.FailureResponseException(from, failure));
}
@Override
public boolean invokeOnFailure()
{
return true;
}
});
return promise;
}
@Override
public <V> void respond(V response, Message<?> message)
{
send(message.responseWith(response), message.respondTo());
super(broadcastAddressAndPort,
Cluster.this::action,
new NetworkDelaySupplier()
{
private final NetworkDelaySupplier delegate = SimulatedMessageDelivery.randomDelay(rs);
@Nullable
@Override
public Duration jitter(Message<?> msg, InetAddressAndPort to)
{
Set<Faults> allowedFaults = allowedMessageFaults.apply(nodes.get(to), msg);
if (!allowedFaults.contains(Faults.DELAY) || broadcastAddressAndPort.equals(to))
return null;
return delegate.jitter(msg, to);
}
},
(to, msg) -> unorderedScheduled.submit(() -> nodes.get(to).handle(msg)),
(action, to, msg) -> logger.warn("{} message {}", action, msg),
unorderedScheduled::schedule,
failures::add);
}
}
@ -1070,7 +893,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
final InetAddressAndPort addressAndPort;
final Collection<Token> tokens;
final ActiveRepairService activeRepairService;
final IVerbHandler verbHandler;
final SimulatedMessageReceiver receiver;
final Messaging messaging;
final IValidationManager validationManager;
private FailingBiConsumer<ColumnFamilyStore, Validator> doValidation = DEFAULT_VALIDATION;
@ -1104,7 +927,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
validator.fail(e);
}
});
this.verbHandler = new IVerbHandler<>()
this.receiver = messaging.receiver(new IVerbHandler<>()
{
private final RepairMessageVerbHandler repairVerbHandler = new RepairMessageVerbHandler(Node.this);
private final IVerbHandler<PaxosStartPrepareCleanup.Request> paxosStartPrepareCleanup = PaxosStartPrepareCleanup.createVerbHandler(Node.this);
@ -1136,7 +959,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
repairVerbHandler.doVerb(message);
}
}
};
});
activeRepairService.start();
}
@ -1176,38 +999,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
}
for (MessageListener l : listeners)
l.preHandle(this, msg);
if (msg.verb().isResponse())
{
// handle callbacks
CallbackKey key = new CallbackKey(msg.id(), msg.from());
if (messaging.callbacks.containsKey(key))
{
CallbackContext callback = messaging.callbacks.remove(key);
if (callback == null)
return;
try
{
if (msg.isFailureResponse())
callback.onFailure(msg.from(), (RequestFailure) msg.payload);
else callback.onResponse(msg);
}
catch (Throwable t)
{
failures.add(t);
}
}
}
else
{
try
{
verbHandler.doVerb(msg);
}
catch (Throwable e)
{
failures.add(e);
}
}
receiver.recieve(msg);
}
public UUID hostId()
@ -1362,6 +1154,12 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
return paxosRepairState;
}
@Override
public Supplier<TimeUUID> timeUUID()
{
return Generators.toGen(Generators.timeUUID()).asSupplier(rs);
}
public String toString()
{
return "Node{" +
@ -1388,14 +1186,6 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
}
}
private static <T> Gen<T> fromQT(org.quicktheories.core.Gen<T> qt)
{
return rs -> {
JavaRandom r = new JavaRandom(rs.asJdkRandom());
return qt.generate(r);
};
}
public static class HackStrat extends LocalStrategy
{
public HackStrat(String keyspaceName, Map<String, String> configOptions)

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

@ -40,6 +40,7 @@ import accord.topology.Shard;
import accord.topology.Topology;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Murmur3Partitioner;
@ -176,7 +177,7 @@ public class AccordConfigurationServiceTest
@Test
public void initialEpochTest() throws Throwable
{
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector());
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
Assert.assertEquals(null, AccordKeyspace.loadEpochDiskState());
service.start();
Assert.assertEquals(null, AccordKeyspace.loadEpochDiskState());
@ -201,7 +202,7 @@ public class AccordConfigurationServiceTest
@Test
public void loadTest() throws Throwable
{
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector());
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
service.start();
Topology topology1 = new Topology(1, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, ID_SET));
@ -221,7 +222,7 @@ public class AccordConfigurationServiceTest
service.reportTopology(topology3);
service.acknowledgeEpoch(EpochReady.done(3), true);
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector());
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
loaded.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
AbstractConfigurationServiceTest.TestListener listener = new AbstractConfigurationServiceTest.TestListener(loaded, true);
loaded.registerListener(listener);
@ -240,7 +241,7 @@ public class AccordConfigurationServiceTest
@Test
public void truncateTest()
{
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector());
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
TestListener serviceListener = new TestListener(service, true);
service.registerListener(serviceListener);
service.start();
@ -258,7 +259,7 @@ public class AccordConfigurationServiceTest
Assert.assertEquals(EpochDiskState.create(3), service.diskState());
serviceListener.assertTruncates(3L);
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector());
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
loaded.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
TestListener loadListener = new TestListener(loaded, true);
loaded.registerListener(loadListener);

4
test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
View File

@ -186,7 +186,7 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
// else this will loop forever...
for (int attempt = 0; attempt < 10; attempt++)
{
TableId tableId = rs.pick(tables.keySet());
TableId tableId = rs.pickOrderedSet(tables.navigableKeySet());
IPartitioner partitioner = tables.get(tableId);
ByteBuffer data = !(partitioner instanceof LocalPartitioner) ? Int32Type.instance.decompose(rs.nextInt())
: fromQT(getTypeSupport(partitioner.getTokenValidator()).bytesGen()).next(rs);
@ -258,7 +258,7 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
for (int i = 0, queries = rs.nextInt(1, 5); i < queries; i++)
{
int store = rs.pick(storesToKeys.keySet());
int store = rs.pickOrderedSet(storesToKeys.navigableKeySet());
var keysForStore = new ArrayList<>(storesToKeys.get(store));
int offset;

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

@ -60,6 +60,9 @@ import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.HeartBeatState;
import org.apache.cassandra.gms.IFailureDetectionEventListener;
import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
@ -94,6 +97,10 @@ public class AccordSyncPropagatorTest
Gen<Ranges> rangesGen = AccordGenerators.ranges().filter(r -> !r.isEmpty());
Gen<List<Node.Id>> nodesGen = Gens.lists(AccordGens.nodes()).unique().ofSizeBetween(1, 40);
qt().withExamples(100).check(rs -> {
// when gossip and cluster metadata don't know an endpoint, retries are avoided (node removed)
// so when instances are created here they are added to gossip to trick the membership check...
Gossiper.instance.clearUnsafe();
List<Node.Id> nodes = nodesGen.next(rs);
Set<Node.Id> nodesAsSet = ImmutableSet.copyOf(nodes);
@ -214,6 +221,7 @@ public class AccordSyncPropagatorTest
Sink sink = new Sink(id);
IFailureDetector fd = new FailureDetector(address);
instances.put(id, new Instace(id, address, cs, sink, fd, cs, new AccordSyncPropagator(id, Cluster.this, sink, fd, scheduler, cs)));
Gossiper.instance.endpointStateMap.put(address, new EndpointState(HeartBeatState.empty()));
}
this.nodeToAddress = nodeToAddress.build();
this.instances = instances.build();

754
test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java Normal file
View File

@ -0,0 +1,754 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumMap;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import com.google.common.collect.Sets;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.ConfigurationService.EpochReady;
import accord.impl.SizeOfIntersectionSorter;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Invariants;
import accord.utils.Property.Command;
import accord.utils.Property.Commands;
import accord.utils.Property.UnitCommand;
import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.SimulatedExecutorFactory;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.gms.IFailureDetectionEventListener;
import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.SimulatedMessageDelivery;
import org.apache.cassandra.net.SimulatedMessageDelivery.Action;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.StubClusterMetadataService;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.DataPlacements;
import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.Pair;
import org.assertj.core.api.Assertions;
import static accord.utils.Property.stateful;
public class EpochSyncTest
{
private static final Logger logger = LoggerFactory.getLogger(EpochSyncTest.class);
static
{
DatabaseDescriptor.clientInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ClusterMetadataService.setInstance(StubClusterMetadataService.forTesting());
}
@Test
public void test()
{
stateful().withExamples(50).check(new Commands<Cluster, Void>()
{
@Override
public Gen<Cluster> genInitialState()
{
return Cluster::new;
}
@Override
public Void createSut(Cluster Cluster)
{
return null;
}
@Override
public Gen<Command<Cluster, Void, ?>> commands(Cluster cluster)
{
List<Node.Id> alive = cluster.alive();
Map<Gen<Command<Cluster, Void, ?>>, Integer> possible = new LinkedHashMap<>();
if (alive.size() < cluster.maxNodes)
{
// add node
possible.put(rs -> {
Node.Id id = new Node.Id(++cluster.nodeCounter);
long token = cluster.tokenGen.nextLong(rs);
while (cluster.tokens.contains(token))
token = cluster.tokenGen.nextLong(rs);
long epoch = cluster.current.epoch.getEpoch() + 1;
long finalToken = token;
return new SimpleCommand("Add Node " + id + "; token=" + token + ", epoch=" + epoch,
c -> c.addNode(id, finalToken));
}, 5);
}
if (alive.size() > cluster.minNodes)
{
possible.put(rs -> {
Node.Id pick = rs.pick(alive);
long token = cluster.instances.get(pick).token;
long epoch = cluster.current.epoch.getEpoch() + 1;
return new SimpleCommand("Remove Node " + pick + "; token=" + token + "; epoch=" + epoch, c -> c.removeNode(pick));
}, 3);
}
if (cluster.hasWork())
{
possible.put(rs -> new SimpleCommand("Process Some",
c -> {//noinspection StatementWithEmptyBody
for (int i = 0, attempts = rs.nextInt(1, 100); i < attempts && c.processOne(); i++)
{
}
}), 10);
}
possible.put(rs -> new SimpleCommand("Validate",
c -> c.validate(false)), 1);
possible.put(rs -> new SimpleCommand("Bump Epoch " + (cluster.current.epoch.getEpoch() + 1),
Cluster::bumpEpoch), 10);
return Gens.oneOf(possible);
}
@Override
public void destroyState(Cluster cluster)
{
cluster.processAll();
cluster.validate(true);
}
});
}
private static class SimpleCommand implements UnitCommand<Cluster, Void>
{
private final String name;
private final Consumer<Cluster> fn;
private SimpleCommand(String name, Consumer<Cluster> fn)
{
this.name = name;
this.fn = fn;
}
@Override
public String detailed(Cluster Cluster)
{
return name;
}
@Override
public void applyUnit(Cluster Cluster)
{
fn.accept(Cluster);
}
@Override
public void runUnit(Void Void)
{
}
}
private static class Cluster
{
private static final int rf = 2;
private static final ReplicationParams replication_params = ReplicationParams.simple(rf);
private static final ReplicationParams meta = ReplicationParams.simpleMeta(1, Collections.singleton("dc1"));
private final RandomSource rs;
private final int minNodes, maxNodes;
private final Gen.LongGen tokenGen;
private final SortedSet<Long> tokens = new TreeSet<>();
private final Map<Node.Id, Instance> instances = new HashMap<>();
private final Set<Node.Id> removed = new HashSet<>();
private final List<Throwable> failures = new ArrayList<>();
private final SimulatedExecutorFactory globalExecutor;
private final ScheduledExecutorPlus scheduler;
private int nodeCounter = 0;
private ClusterMetadata current = new ClusterMetadata(Murmur3Partitioner.instance, Directory.EMPTY,
new DistributedSchema(Keyspaces.of(
DistributedMetadataLogKeyspace.initialMetadata(Collections.singleton("dc1")),
KeyspaceMetadata.create("test", KeyspaceParams.simple(rf), Tables.of(TableMetadata.minimal("test", "tb1").unbuild().params(TableParams.builder().transactionalMode(TransactionalMode.full).build()).build())))));
private final IFailureDetector fd = new IFailureDetector()
{
@Override
public boolean isAlive(InetAddressAndPort ep)
{
return !removed.contains(nodeId(ep));
}
@Override
public void interpret(InetAddressAndPort ep)
{
}
@Override
public void report(InetAddressAndPort ep)
{
}
@Override
public void remove(InetAddressAndPort ep)
{
}
@Override
public void forceConviction(InetAddressAndPort ep)
{
}
@Override
public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener)
{
}
@Override
public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener)
{
}
};
private static InetAddressAndPort address(Node.Id id)
{
try
{
return InetAddressAndPort.getByAddress(ByteArrayUtil.bytes(id.id));
}
catch (UnknownHostException e)
{
throw new AssertionError("Unable to create address for id " + id, e);
}
}
public enum EpochTracker { topologyManager, accordSyncPropagator, configurationService}
Set<EpochTracker> globalSynced(long epoch)
{
return alive().stream()
.filter(n -> instances.get(n).epoch.getEpoch() <= epoch)
.map(n -> instances.get(n).synced(epoch))
.reduce(EnumSet.allOf(EpochTracker.class), Sets::intersection);
}
boolean allSynced(long epoch)
{
Set<EpochTracker> done = globalSynced(epoch);
return done.contains(EpochTracker.topologyManager);
}
private static Node.Id nodeId(InetAddressAndPort address)
{
return new Node.Id(ByteArrayUtil.getInt(address.addressBytes));
}
public Cluster(RandomSource rs)
{
this.rs = rs;
this.minNodes = 3;
this.maxNodes = 10;
this.tokenGen = rs2 -> rs2.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
this.globalExecutor = new SimulatedExecutorFactory(accord.utilsfork.RandomSource.wrap(rs.asJdkRandom()), failures::add);
this.scheduler = globalExecutor.scheduled("ignored");
Stage.MISC.unsafeSetExecutor(scheduler);
scheduler.scheduleWithFixedDelay(() -> {
if (aliveCount() < 2) return;
if (!partitions.isEmpty() && rs.nextBoolean())
{
// remove partition
if (partitions.size() == 1)
{
partitions.clear();
return;
}
partitions.remove(rs.pickOrderedSet(partitions));
}
else
{
// add partition
List<Node.Id> alive = alive();
InetAddressAndPort a = address(rs.pick(alive));
InetAddressAndPort b = address(rs.pick(alive));
while (a.equals(b))
b = address(rs.pick(alive));
partitions.add(new Connection(a, b));
}
}, 1, 1, TimeUnit.MINUTES);
}
void validate(boolean isDone)
{
for (Node.Id id : alive())
{
Instance inst = instances.get(id);
if (removed.contains(id)) continue; // ignore removed nodes
AccordConfigurationService conf = inst.config;
TopologyManager tm = inst.topology;
for (long epoch = inst.epoch.getEpoch(); epoch <= current.epoch.getEpoch(); epoch++)
{
// validate config
EpochSnapshot snapshot = conf.getEpochSnapshot(epoch);
if (isDone)
{
Assertions.assertThat(snapshot).describedAs("node%s does not have epoch %d", id, epoch).isNotNull();
Assertions.assertThat(snapshot.syncStatus).isEqualTo(AccordConfigurationService.SyncStatus.COMPLETED);
// validate topology manager
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Ranges ranges = tm.globalForEpoch(epoch).ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
Assertions.assertThat(actual).describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.subtract(actual)).isEqualTo(ranges);
}
else
{
if (snapshot == null || snapshot.syncStatus != AccordConfigurationService.SyncStatus.COMPLETED) continue;
if (!allSynced(epoch))
continue;
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Topology topology = tm.globalForEpoch(epoch);
Ranges ranges = topology.ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
// TopologyManager defines syncComplete for an epoch as (epoch - 1).syncComplete. This means that an epoch has reached quorum, but will still miss ranges as previous epochs have not
if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.syncComplete(epoch - 1).mergeTouching()))
continue;
Assertions.assertThat(actual)
.describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", id, epoch, ranges.subtract(actual), topology.nodes(),
LongStream.range(inst.epoch.getEpoch(), epoch + 1).mapToObj(e -> e + " -> " + conf.getEpochSnapshot(e).syncStatus + "(synced=" + globalSynced(e) + "): " + tm.syncComplete(e)).collect(Collectors.joining("\n")))
.isEqualTo(ranges);
}
}
}
}
String displayTopology()
{
List<Node.Id> alive = alive();
List<Pair<Node.Id, Long>> withToken = new ArrayList<>(alive.size());
for (Node.Id n : alive)
withToken.add(Pair.create(n, instances.get(n).token));
withToken.sort(Comparator.comparing(a -> a.right));
StringBuilder sb = new StringBuilder();
for (var p : withToken)
sb.append(p.left).append('\t').append(p.right).append('\n');
return sb.toString();
}
@Override
public String toString()
{
return "Topology:\n" + displayTopology();
}
boolean hasWork()
{
return globalExecutor.hasWork();
}
boolean processOne()
{
boolean result = globalExecutor.processOne();
checkFailures();
return result;
}
@SuppressWarnings("StatementWithEmptyBody")
void processAll()
{
while (processOne())
{
}
}
public void checkFailures()
{
if (Thread.interrupted())
failures.add(new InterruptedException());
if (failures.isEmpty()) return;
AssertionError error = new AssertionError("Unexpected exceptions found");
failures.forEach(error::addSuppressed);
failures.clear();
throw error;
}
List<Node.Id> alive()
{
ArrayList<Node.Id> ids = new ArrayList<>(Sets.difference(instances.keySet(), removed));
ids.sort(Comparator.naturalOrder());
return ids;
}
int aliveCount()
{
return instances.size() - removed.size();
}
private final NavigableSet<Connection> partitions = new TreeSet<>();
private boolean partitioned(InetAddressAndPort self, InetAddressAndPort to)
{
return partitions.contains(new Connection(self, to));
}
private SimulatedMessageDelivery createMessaging(Node.Id id)
{
InetAddressAndPort address = address(id);
return new SimulatedMessageDelivery(address,
(self, msg, to) -> {
if (removed.contains(nodeId(self)) || removed.contains(nodeId(to)))
return Action.DROP;
if (!self.equals(to) && partitioned(self, to))
return Action.DROP_PARTITIONED;
if (rs.decide(.01))
return rs.nextBoolean() ? Action.DELIVER_WITH_FAILURE : Action.FAILURE;
return Action.DELIVER;
},
SimulatedMessageDelivery.randomDelay(accord.utilsfork.RandomSource.wrap(rs.asJdkRandom())),
(to, msg) -> instances.get(nodeId(to)).reciver.recieve(msg),
(action, to, msg) -> logger.warn("{} message {}", action, msg),
scheduler::schedule,
failures::add);
}
void addNode(Node.Id id, long token)
{
Invariants.checkState(!tokens.contains(token), "Attempted to add token %d for node %s but token is already taken", token, id);
Epoch epoch = Epoch.create(current.epoch.getEpoch() + 1);
Instance instance = new Instance(id, token, epoch, createMessaging(id), fd);
instances.put(id, instance);
tokens.add(token);
current = current.forceEpoch(epoch)
.withPlacements(DataPlacements.builder(2)
.with(meta, DataPlacement.empty())
.with(replication_params, rebuildPlacements(epoch))
.build())
.withDirectory(current.directory.with(new NodeAddresses(address(id)), new Location("dc1", "r1")));
notify(current);
}
void removeNode(Node.Id pick)
{
Instance inst = Objects.requireNonNull(instances.get(pick), "Unknown id " + pick);
Invariants.checkState(!removed.contains(pick), "Can not remove node twice; node " + pick);
tokens.remove(inst.token);
removed.add(pick);
inst.stop();
current = current.forceEpoch(Epoch.create(current.epoch.getEpoch() + 1))
.withDirectory(current.directory.without(new NodeId(pick.id)));
current = current.withPlacements(DataPlacements.builder(2)
.with(meta, DataPlacement.empty())
.with(replication_params, rebuildPlacements(current.epoch))
.build());
notify(current);
}
private DataPlacement rebuildPlacements(Epoch epoch)
{
DataPlacement.Builder builder = DataPlacement.builder();
for (Node.Id inst : alive())
for (Replica replica : instances.get(inst).replica())
builder.withReadReplica(epoch, replica).withWriteReplica(epoch, replica);
return builder.build();
}
void bumpEpoch()
{
current = current.forceEpoch(Epoch.create(current.epoch.getEpoch() + 1));
notify(current);
}
private void notify(ClusterMetadata current)
{
Ranges ranges = AccordTopology.createAccordTopology(current).ranges().mergeTouching();
if (!current.directory.isEmpty())
Assertions.assertThat(ranges).hasSize(1);
((StubClusterMetadataService) ClusterMetadataService.instance()).setMetadata(current);
for (Node.Id id : alive())
{
Instance inst = instances.get(id);
inst.maybeStart();
inst.config.maybeReportMetadata(current);
}
}
@SuppressWarnings("SameParameterValue")
private <T> AsyncChain<T> schedule(long time, TimeUnit unit, Callable<T> task)
{
return new AsyncChains.Head<>()
{
@Override
protected void start(BiConsumer<? super T, Throwable> callback)
{
scheduler.schedule(() -> {
T value;
try
{
value = task.call();
}
catch (Throwable t)
{
callback.accept(null, t);
return;
}
callback.accept(value, null);
}, time, unit);
}
};
}
private enum Status { Init, Started}
private class Instance
{
private final Node.Id id;
private final long token;
private final AccordConfigurationService config;
private final SimulatedMessageDelivery messaging;
private final SimulatedMessageDelivery.SimulatedMessageReceiver reciver;
private final TopologyManager topology;
private final Epoch epoch;
private Status status = Status.Init;
Instance(Node.Id node, long token, Epoch epoch, SimulatedMessageDelivery messagingService, IFailureDetector failureDetector)
{
this.id = node;
this.token = token;
this.epoch = epoch;
this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, id);
AccordConfigurationService.DiskStateManager instance = MockDiskStateManager.instance;
config = new AccordConfigurationService(node, messagingService, failureDetector, instance, scheduler);
config.registerListener(new ConfigurationService.Listener()
{
@Override
public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean startSync)
{
// EpochReady ready = EpochReady.done(topology.epoch());
AsyncResult<Void> metadata = schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult();
AsyncResult<Void> coordination = metadata.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null)).beginAsResult();
AsyncResult<Void> data = coordination.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null)).beginAsResult();
AsyncResult<Void> reads = data.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null)).beginAsResult();
EpochReady ready = new EpochReady(topology.epoch(), metadata, coordination, data, reads);
topology().onTopologyUpdate(topology, () -> ready);
ready.coordination.addCallback(() -> topology().onEpochSyncComplete(id, topology.epoch()));
if (topology().minEpoch() == topology.epoch() && topology().epoch() != topology.epoch())
return ready.coordination;
config.acknowledgeEpoch(ready, startSync);
return ready.coordination;
}
@Override
public void onRemoteSyncComplete(Node.Id node, long epoch)
{
topology.onEpochSyncComplete(node, epoch);
}
@Override
public void onRemoveNodes(long epoch, Collection<Node.Id> removed)
{
topology.onRemoveNodes(epoch, removed);
}
@Override
public void truncateTopologyUntil(long epoch)
{
topology.truncateTopologyUntil(epoch);
}
@Override
public void onEpochClosed(Ranges ranges, long epoch)
{
topology.onEpochClosed(ranges, epoch);
}
@Override
public void onEpochRedundant(Ranges ranges, long epoch)
{
topology.onEpochRedundant(ranges, epoch);
}
});
Map<Verb, IVerbHandler<?>> handlers = new EnumMap<>(Verb.class);
//noinspection unchecked
handlers.put(Verb.ACCORD_SYNC_NOTIFY_REQ, msg -> AccordService.receive(messagingService, config, (Message<List<AccordSyncPropagator.Notification>>) (Message<?>) msg));
this.messaging = messagingService;
this.reciver = messagingService.receiver(new SimulatedMessageDelivery.SimpleVerbHandler(handlers));
}
void maybeStart()
{
if (status == Status.Init)
{
start();
status = Status.Started;
}
}
private void start()
{
config.start();
}
TopologyManager topology()
{
return topology;
}
Collection<Replica> replica()
{
InetAddressAndPort address = Cluster.address(id);
SortedSet<Long> lessThan = tokens.headSet(token);
if (lessThan.isEmpty())
{
// wrap around
return Arrays.asList(new Replica(address, new LongToken(Long.MIN_VALUE), new LongToken(token), true),
new Replica(address, new LongToken(tokens.last()), new LongToken(Long.MIN_VALUE), true));
}
return Collections.singletonList(new Replica(address, new LongToken(lessThan.last()), new LongToken(token), true));
}
Set<EpochTracker> synced(long epoch)
{
if (epoch < this.epoch.getEpoch()) throw new IllegalArgumentException("Asked for epoch before this instance existed");
EnumSet<EpochTracker> done = EnumSet.noneOf(EpochTracker.class);
EpochSnapshot snapshot = config.getEpochSnapshot(epoch);
if (snapshot != null && snapshot.syncStatus == AccordConfigurationService.SyncStatus.COMPLETED)
done.add(EpochTracker.configurationService);
if (topology.hasReachedQuorum(epoch))
done.add(EpochTracker.topologyManager);
if (!config.syncPropagator().hasPending(epoch))
done.add(EpochTracker.accordSyncPropagator);
return done;
}
void stop()
{
messaging.stop();
}
}
}
private static class Connection implements Comparable<Connection>
{
final InetAddressAndPort from, to;
private Connection(InetAddressAndPort from, InetAddressAndPort to)
{
this.from = from;
this.to = to;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Connection that = (Connection) o;
return from.equals(that.from) && to.equals(that.to);
}
@Override
public int hashCode()
{
return Objects.hash(from, to);
}
@Override
public String toString()
{
return "Connection{" + "from=" + from + ", to=" + to + '}';
}
@Override
public int compareTo(Connection o)
{
int rc = from.compareTo(o.from);
if (rc == 0)
rc = to.compareTo(o.to);
return rc;
}
}
}

93
test/unit/org/apache/cassandra/service/accord/LoggingDiskStateManager.java Normal file
View File

@ -0,0 +1,93 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Topology;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Set;
/**
* When trying to inspect the order in which disk state is modified, this class can aid by adding logging. This class
* mostly exists for testing to aid in debugging.
*/
@SuppressWarnings("unused")
@VisibleForTesting
public class LoggingDiskStateManager implements AccordConfigurationService.DiskStateManager {
private static final Logger logger = LoggerFactory.getLogger(LoggingDiskStateManager.class);
private final Node.Id self;
private final AccordConfigurationService.DiskStateManager delegate;
public LoggingDiskStateManager(Node.Id self, AccordConfigurationService.DiskStateManager delegate) {
this.self = self;
this.delegate = delegate;
}
@Override
public AccordKeyspace.EpochDiskState loadTopologies(AccordKeyspace.TopologyLoadConsumer consumer) {
logger.info("[node={}] Calling loadTopologies()", self);
return delegate.loadTopologies(consumer);
}
@Override
public AccordKeyspace.EpochDiskState setNotifyingLocalSync(long epoch, Set<Node.Id> pending, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling setNotifyingLocalSync({}, {}, {})", self, epoch, pending, diskState);
return delegate.setNotifyingLocalSync(epoch, pending, diskState);
}
@Override
public AccordKeyspace.EpochDiskState setCompletedLocalSync(long epoch, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling setCompletedLocalSync({}, {})", self, epoch, diskState);
return delegate.setCompletedLocalSync(epoch, diskState);
}
@Override
public AccordKeyspace.EpochDiskState markLocalSyncAck(Node.Id id, long epoch, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling markLocalSyncAck({}, {}, {})", self, id, epoch, diskState);
return delegate.markLocalSyncAck(id, epoch, diskState);
}
@Override
public AccordKeyspace.EpochDiskState saveTopology(Topology topology, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling saveTopology({}, {})", self, topology.epoch(), diskState);
return delegate.saveTopology(topology, diskState);
}
@Override
public AccordKeyspace.EpochDiskState markRemoteTopologySync(Node.Id id, long epoch, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling markRemoteTopologySync({}, {}, {})", self, id, epoch, diskState);
return delegate.markRemoteTopologySync(id, epoch, diskState);
}
@Override
public AccordKeyspace.EpochDiskState markClosed(Ranges ranges, long epoch, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling markClosed({}, {}, {})", self, ranges, epoch, diskState);
return delegate.markClosed(ranges, epoch, diskState);
}
@Override
public AccordKeyspace.EpochDiskState truncateTopologyUntil(long epoch, AccordKeyspace.EpochDiskState diskState) {
logger.info("[node={}] Calling truncateTopologyUntil({}, {})", self, epoch, diskState);
return delegate.truncateTopologyUntil(epoch, diskState);
}
}

79
test/unit/org/apache/cassandra/service/accord/MockDiskStateManager.java Normal file
View File

@ -0,0 +1,79 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.utils.Invariants;
import java.util.Set;
public enum MockDiskStateManager implements AccordConfigurationService.DiskStateManager {
instance;
@Override
public AccordKeyspace.EpochDiskState loadTopologies(AccordKeyspace.TopologyLoadConsumer consumer) {
return AccordKeyspace.EpochDiskState.EMPTY;
}
@Override
public AccordKeyspace.EpochDiskState setNotifyingLocalSync(long epoch, Set<Node.Id> pending, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
@Override
public AccordKeyspace.EpochDiskState setCompletedLocalSync(long epoch, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
@Override
public AccordKeyspace.EpochDiskState markLocalSyncAck(Node.Id id, long epoch, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
@Override
public AccordKeyspace.EpochDiskState saveTopology(Topology topology, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, topology.epoch());
}
@Override
public AccordKeyspace.EpochDiskState markRemoteTopologySync(Node.Id node, long epoch, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
@Override
public AccordKeyspace.EpochDiskState markClosed(Ranges ranges, long epoch, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
@Override
public AccordKeyspace.EpochDiskState truncateTopologyUntil(long epoch, AccordKeyspace.EpochDiskState diskState) {
return maybeUpdateMaxEpoch(diskState, epoch);
}
private static AccordKeyspace.EpochDiskState maybeUpdateMaxEpoch(AccordKeyspace.EpochDiskState diskState, long epoch) {
if (diskState.isEmpty())
return AccordKeyspace.EpochDiskState.create(epoch);
Invariants.checkArgument(epoch >= diskState.minEpoch, "Epoch %d < %d (min)", epoch, diskState.minEpoch);
if (epoch > diskState.maxEpoch)
diskState = diskState.withNewMaxEpoch(epoch);
return diskState;
}
}

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

@ -21,7 +21,7 @@ package org.apache.cassandra.utils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
@ -99,7 +99,7 @@ public class StatefulRangeTreeTest
@Override
public Gen<Command<State, Sut, ?>> commands(State state)
{
Map<Gen<Command<State, Sut, ?>>, Integer> possible = new HashMap<>();
Map<Gen<Command<State, Sut, ?>>, Integer> possible = new LinkedHashMap<>();
possible.put(rs -> new Create(state.newRange(rs), SMALL_INT_GEN.nextInt(rs)), state.createWeight);
possible.put(rs -> new Read(state.newRange(rs)), state.readWeight);
possible.put(rs -> new KeyRead(IntKey.routing(state.tokenGen.nextInt(rs))), state.readWeight);
@ -108,15 +108,15 @@ public class StatefulRangeTreeTest
possible.put(ignore -> Clear.instance, state.clearWeight);
if (!state.uniqRanges.isEmpty())
{
possible.put(rs -> new Read(rs.pick(state.uniqRanges)), state.readWeight);
possible.put(rs -> new Read(rs.pickOrderedSet(state.uniqRanges)), state.readWeight);
possible.put(rs -> {
Range range = rs.pick(state.uniqRanges);
Range range = rs.pickOrderedSet(state.uniqRanges);
int token = rs.nextInt(((IntKey.Routing) range.start()).key, ((IntKey.Routing) range.end()).key) + 1;
return new KeyRead(IntKey.routing(token));
}, state.readWeight);
possible.put(rs -> new RangeRead(rs.pick(state.uniqRanges)), state.readWeight);
possible.put(rs -> new Update(rs.pick(state.uniqRanges), SMALL_INT_GEN.nextInt(rs)), state.updateWeight);
possible.put(rs -> new Delete(rs.pick(state.uniqRanges)), state.deleteWeight);
possible.put(rs -> new RangeRead(rs.pickOrderedSet(state.uniqRanges)), state.readWeight);
possible.put(rs -> new Update(rs.pickOrderedSet(state.uniqRanges), SMALL_INT_GEN.nextInt(rs)), state.updateWeight);
possible.put(rs -> new Delete(rs.pickOrderedSet(state.uniqRanges)), state.deleteWeight);
}
return Gens.oneOf(possible);
}